A Share MCP

Instructions

Implementation Reference

• src/tools/financial_reports.py:45-64 (handler)

  The primary MCP tool handler for 'get_operation_data', decorated with @app.tool() for automatic registration. It delegates to call_financial_data_tool which handles validation, data fetching from the active data source, error handling, and markdown formatting.
  @app.tool() def get_operation_data(code: str, year: str, quarter: int, limit: int = 250, format: str = "markdown") -> str: """ Get quarterly operation capability data (e.g., turnover ratios) for a stock. Args: code: The stock code (e.g., 'sh.600000'). year: The 4-digit year (e.g., '2023'). quarter: The quarter (1, 2, 3, or 4). Returns: Operation capability metrics table. """ return call_financial_data_tool( "get_operation_data", active_data_source.get_operation_data, "Operation Capability", code, year, quarter, limit=limit, format=format )
• src/tools/base.py:15-73 (helper)

  Shared helper invoked by the handler to validate inputs, call the data source method (active_data_source.get_operation_data), handle exceptions, and format the DataFrame output as markdown table.
  def call_financial_data_tool( tool_name: str, # Pass the bound method like active_data_source.get_profit_data data_source_method: Callable, data_type_name: str, code: str, year: str, quarter: int, *, limit: int = 250, format: str = "markdown", ) -> str: """ Helper function to reduce repetition for financial data tools Args: tool_name: Name of the tool for logging data_source_method: Method to call on the data source data_type_name: Type of financial data (for logging) code: Stock code year: Year to query quarter: Quarter to query Returns: Markdown formatted string with results or error message """ logger.info(f"Tool '{tool_name}' called for {code}, {year}Q{quarter}") try: # Basic validation if not year.isdigit() or len(year) != 4: logger.warning(f"Invalid year format requested: {year}") return f"Error: Invalid year '{year}'. Please provide a 4-digit year." if not 1 <= quarter <= 4: logger.warning(f"Invalid quarter requested: {quarter}") return f"Error: Invalid quarter '{quarter}'. Must be between 1 and 4." # Call the appropriate method on the already instantiated active_data_source df = data_source_method(code=code, year=year, quarter=quarter) logger.info( f"Successfully retrieved {data_type_name} data for {code}, {year}Q{quarter}.") meta = {"code": code, "year": year, "quarter": quarter, "dataset": data_type_name} return format_table_output(df, format=format, max_rows=limit, meta=meta) except NoDataFoundError as e: logger.warning(f"NoDataFoundError for {code}, {year}Q{quarter}: {e}") return f"Error: {e}" except LoginError as e: logger.error(f"LoginError for {code}: {e}") return f"Error: Could not connect to data source. {e}" except DataSourceError as e: logger.error(f"DataSourceError for {code}: {e}") return f"Error: An error occurred while fetching data. {e}" except ValueError as e: logger.warning(f"ValueError processing request for {code}: {e}") return f"Error: Invalid input parameter. {e}" except Exception as e: logger.exception( f"Unexpected Exception processing {tool_name} for {code}: {e}") return f"Error: An unexpected error occurred: {e}"
• src/baostock_data_source.py:417-419 (helper)

  Implementation of get_operation_data in BaostockDataSource class, which calls Baostock's query_operation_data via the shared _fetch_financial_data helper to retrieve raw DataFrame.
  def get_operation_data(self, code: str, year: str, quarter: int) -> pd.DataFrame: """Fetches quarterly operation capability data using Baostock.""" return _fetch_financial_data(bs.query_operation_data, "Operation Capability", code, year, quarter)
• src/baostock_data_source.py:26-74 (helper)

  Internal helper in BaostockDataSource that performs the actual Baostock API query for financial data, handles login context, error checking, and constructs the DataFrame from results.
  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 except (LoginError, NoDataFoundError, DataSourceError, ValueError) as e: logger.warning( f"Caught known error fetching {data_type_name} data for {code}: {type(e).__name__}") raise e except Exception as e: logger.exception( f"Unexpected error fetching {data_type_name} data for {code}: {e}") raise DataSourceError( f"Unexpected error fetching {data_type_name} data for {code}: {e}")
• src/tools/financial_reports.py:45-64 (registration)

  The @app.tool() decorator registers the get_operation_data function as an MCP tool within the FastMCP app.
  @app.tool() def get_operation_data(code: str, year: str, quarter: int, limit: int = 250, format: str = "markdown") -> str: """ Get quarterly operation capability data (e.g., turnover ratios) for a stock. Args: code: The stock code (e.g., 'sh.600000'). year: The 4-digit year (e.g., '2023'). quarter: The quarter (1, 2, 3, or 4). Returns: Operation capability metrics table. """ return call_financial_data_tool( "get_operation_data", active_data_source.get_operation_data, "Operation Capability", code, year, quarter, limit=limit, format=format )