A-Share MCP Server

Instructions

Implementation Reference

• src/tools/stock_market.py:25-110 (handler)

  Primary MCP tool handler for get_historical_k_data. Validates parameters, fetches data from active_data_source, formats output using format_table_output, and handles various exceptions.
  def get_historical_k_data( code: str, start_date: str, end_date: str, frequency: str = "d", adjust_flag: str = "3", fields: Optional[List[str]] = None, limit: int = 250, format: str = "markdown", ) -> str: """ Fetches historical K-line (OHLCV) data for a Chinese A-share stock. Args: code: The stock code in Baostock format (e.g., 'sh.600000', 'sz.000001'). start_date: Start date in 'YYYY-MM-DD' format. end_date: End date in 'YYYY-MM-DD' format. frequency: Data frequency. Valid options (from Baostock): 'd': daily 'w': weekly 'm': monthly '5': 5 minutes '15': 15 minutes '30': 30 minutes '60': 60 minutes Defaults to 'd'. adjust_flag: Adjustment flag for price/volume. Valid options (from Baostock): '1': Forward adjusted (后复权) '2': Backward adjusted (前复权) '3': Non-adjusted (不复权) Defaults to '3'. fields: Optional list of specific data fields to retrieve (must be valid Baostock fields). If None or empty, default fields will be used (e.g., date, code, open, high, low, close, volume, amount, pctChg). limit: Max rows to return. Defaults to 250. format: Output format: 'markdown' | 'json' | 'csv'. Defaults to 'markdown'. Returns: A Markdown formatted string containing the K-line data table, or an error message. The table might be truncated if the result set is too large. """ logger.info( f"Tool 'get_historical_k_data' called for {code} ({start_date}-{end_date}, freq={frequency}, adj={adjust_flag}, fields={fields})") try: # Validate frequency and adjust_flag if necessary (basic example) valid_freqs = ['d', 'w', 'm', '5', '15', '30', '60'] valid_adjusts = ['1', '2', '3'] if frequency not in valid_freqs: logger.warning(f"Invalid frequency requested: {frequency}") return f"Error: Invalid frequency '{frequency}'. Valid options are: {valid_freqs}" if adjust_flag not in valid_adjusts: logger.warning(f"Invalid adjust_flag requested: {adjust_flag}") return f"Error: Invalid adjust_flag '{adjust_flag}'. Valid options are: {valid_adjusts}" # Call the injected data source df = active_data_source.get_historical_k_data( code=code, start_date=start_date, end_date=end_date, frequency=frequency, adjust_flag=adjust_flag, fields=fields, ) # Format the result logger.info( f"Successfully retrieved K-data for {code}, formatting output.") meta = {"code": code, "start_date": start_date, "end_date": end_date, "frequency": frequency, "adjust_flag": adjust_flag} return format_table_output(df, format=format, max_rows=limit, meta=meta) except NoDataFoundError as e: logger.warning(f"NoDataFoundError for {code}: {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: # Catch-all for unexpected errors logger.exception( f"Unexpected Exception processing get_historical_k_data for {code}: {e}") return f"Error: An unexpected error occurred: {e}"
• mcp_server.py:51-51 (registration)

  Explicit registration of stock market tools (including get_historical_k_data) by calling register_stock_market_tools on the FastMCP app instance with the active data source.
  register_stock_market_tools(app, active_data_source)
• src/data_source_interface.py:28-65 (schema)

  Abstract method in FinancialDataSource interface defining the input schema, output type (pd.DataFrame), and expected behavior for get_historical_k_data implementations.
  @abstractmethod def get_historical_k_data( self, code: str, start_date: str, end_date: str, frequency: str = "d", adjust_flag: str = "3", fields: Optional[List[str]] = None, ) -> pd.DataFrame: """ Fetches historical K-line (OHLCV) data for a given stock code. Args: code: The stock code (e.g., 'sh.600000', 'sz.000001'). start_date: Start date in 'YYYY-MM-DD' format. end_date: End date in 'YYYY-MM-DD' format. frequency: Data frequency. Common values depend on the underlying source (e.g., 'd' for daily, 'w' for weekly, 'm' for monthly, '5', '15', '30', '60' for minutes). Defaults to 'd'. adjust_flag: Adjustment flag for historical data. Common values depend on the source (e.g., '1' for forward adjusted, '2' for backward adjusted, '3' for non-adjusted). Defaults to '3'. fields: Optional list of specific fields to retrieve. If None, retrieves default fields defined by the implementation. Returns: A pandas DataFrame containing the historical K-line data, with columns corresponding to the requested fields. 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 input parameters are invalid. """ pass
• src/baostock_data_source.py:196-261 (helper)

  Concrete implementation of get_historical_k_data in BaostockDataSource, using bs.query_history_k_data_plus API, handling Baostock-specific errors, and returning a DataFrame.
  def get_historical_k_data( self, code: str, start_date: str, end_date: str, frequency: str = "d", adjust_flag: str = "3", fields: Optional[List[str]] = None, ) -> pd.DataFrame: """Fetches historical K-line data using Baostock.""" logger.info( f"Fetching K-data for {code} ({start_date} to {end_date}), freq={frequency}, adjust={adjust_flag}") try: formatted_fields = self._format_fields(fields, DEFAULT_K_FIELDS) logger.debug( f"Requesting fields from Baostock: {formatted_fields}") with baostock_login_context(): rs = bs.query_history_k_data_plus( code, formatted_fields, start_date=start_date, end_date=end_date, frequency=frequency, adjustflag=adjust_flag ) if rs.error_code != '0': logger.error( f"Baostock API error (K-data) for {code}: {rs.error_msg} (code: {rs.error_code})") # Check common error codes, e.g., for no data if "no record found" in rs.error_msg.lower() or rs.error_code == '10002': # Example error code raise NoDataFoundError( f"No historical data found for {code} in the specified range. Baostock msg: {rs.error_msg}") else: raise DataSourceError( f"Baostock API error fetching K-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 historical data found for {code} in range (empty result set from Baostock).") raise NoDataFoundError( f"No historical data found for {code} in the specified range (empty result set).") # Crucial: Use rs.fields for column names result_df = pd.DataFrame(data_list, columns=rs.fields) logger.info(f"Retrieved {len(result_df)} records for {code}.") return result_df except (LoginError, NoDataFoundError, DataSourceError, ValueError) as e: # Re-raise known errors logger.warning( f"Caught known error fetching K-data for {code}: {type(e).__name__}") raise e except Exception as e: # Wrap unexpected errors # Use logger.exception to include traceback logger.exception( f"Unexpected error fetching K-data for {code}: {e}") raise DataSourceError( f"Unexpected error fetching K-data for {code}: {e}")
• src/tools/stock_market.py:15-23 (registration)

  Registration function that defines and registers the get_historical_k_data tool (and others) using @app.tool() decorator on the FastMCP app.
  def register_stock_market_tools(app: FastMCP, active_data_source: FinancialDataSource): """ Register stock market data tools with the MCP app. Args: app: The FastMCP app instance active_data_source: The active financial data source """