A Share MCP

Instructions

Implementation Reference

• src/tools/stock_market.py:30-87 (handler)

  MCP tool handler for 'get_historical_k_data'. Decorated with @app.tool(), performs logging, and delegates execution to the use case function via run_tool_with_handling for error handling.
  @app.tool() 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})" ) return run_tool_with_handling( lambda: fetch_historical_k_data( active_data_source, code=code, start_date=start_date, end_date=end_date, frequency=frequency, adjust_flag=adjust_flag, fields=fields, limit=limit, format=format, ), context=f"get_historical_k_data:{code}", )
• src/data_source_interface.py:28-65 (schema)

  Abstract interface definition in FinancialDataSource specifying the exact input parameters, types, and return type (pd.DataFrame) for the historical K data method, serving as the schema for 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
• mcp_server.py:51-51 (registration)

  Explicit call to register_stock_market_tools during app initialization, which defines and registers the get_historical_k_data tool with the FastMCP app instance.
  register_stock_market_tools(app, active_data_source)
• src/use_cases/stock_market.py:17-49 (helper)

  Use case orchestrator that validates inputs using dedicated validators, fetches raw data from the FinancialDataSource, adds metadata, and formats the output as markdown/JSON/CSV.
  def fetch_historical_k_data( data_source: FinancialDataSource, *, 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: validate_frequency(frequency) validate_adjust_flag(adjust_flag) validate_output_format(format) df = data_source.get_historical_k_data( code=code, start_date=start_date, end_date=end_date, frequency=frequency, adjust_flag=adjust_flag, fields=fields, ) 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)
• src/baostock_data_source.py:196-261 (helper)

  Core data fetching implementation in BaostockDataSource class. Queries the Baostock API using query_history_k_data_plus, handles errors, login context, and returns a pandas 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}")