A-Share MCP Server

Instructions

Implementation Reference

• src/tools/stock_market.py:111-154 (handler)

  The primary handler function for the 'get_stock_basic_info' tool, decorated with @app.tool(). It fetches data from the active FinancialDataSource, handles errors, and formats the output as markdown, JSON, or CSV.
  @app.tool() def get_stock_basic_info(code: str, fields: Optional[List[str]] = None, format: str = "markdown") -> str: """ Fetches basic information for a given Chinese A-share stock. Args: code: The stock code in Baostock format (e.g., 'sh.600000', 'sz.000001'). fields: Optional list to select specific columns from the available basic info (e.g., ['code', 'code_name', 'industry', 'listingDate']). If None or empty, returns all available basic info columns from Baostock. Returns: Basic stock information in the requested format. """ logger.info( f"Tool 'get_stock_basic_info' called for {code} (fields={fields})") try: # Call the injected data source # Pass fields along; BaostockDataSource implementation handles selection df = active_data_source.get_stock_basic_info( code=code, fields=fields) # Format the result (basic info usually small) logger.info( f"Successfully retrieved basic info for {code}, formatting output.") meta = {"code": code} return format_table_output(df, format=format, max_rows=df.shape[0] if df is not None else 0, 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 or requested field not available. {e}" except Exception as e: logger.exception( f"Unexpected Exception processing get_stock_basic_info for {code}: {e}") return f"Error: An unexpected error occurred: {e}"
• mcp_server.py:51-51 (registration)

  The call to register_stock_market_tools in the main MCP server setup, which registers the get_stock_basic_info tool (and others) with the FastMCP app instance.
  register_stock_market_tools(app, active_data_source)
• src/baostock_data_source.py:262-323 (helper)

  The helper method in BaostockDataSource that implements the data fetching logic for stock basic info using Baostock's query_stock_basic API, including field selection and error handling.
  def get_stock_basic_info(self, code: str, fields: Optional[List[str]] = None) -> pd.DataFrame: """Fetches basic stock information using Baostock.""" logger.info(f"Fetching basic info for {code}") try: # Note: query_stock_basic doesn't seem to have a fields parameter in docs, # but we keep the signature consistent. It returns a fixed set. # We will use the `fields` argument post-query to select columns if needed. logger.debug( f"Requesting basic info for {code}. Optional fields requested: {fields}") with baostock_login_context(): # Example: Fetch basic info; adjust API call if needed based on baostock docs # rs = bs.query_stock_basic(code=code, code_name=code_name) # If supporting name lookup rs = bs.query_stock_basic(code=code) if rs.error_code != '0': logger.error( f"Baostock API error (Basic Info) 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 basic info found for {code}. Baostock msg: {rs.error_msg}") else: raise DataSourceError( f"Baostock API error fetching basic info: {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 basic info found for {code} (empty result set from Baostock).") raise NoDataFoundError( f"No basic info found for {code} (empty result set).") # Crucial: Use rs.fields for column names result_df = pd.DataFrame(data_list, columns=rs.fields) logger.info( f"Retrieved basic info for {code}. Columns: {result_df.columns.tolist()}") # Optional: Select subset of columns if `fields` argument was provided if fields: available_cols = [ col for col in fields if col in result_df.columns] if not available_cols: raise ValueError( f"None of the requested fields {fields} are available in the basic info result.") logger.debug( f"Selecting columns: {available_cols} from basic info for {code}") result_df = result_df[available_cols] return result_df except (LoginError, NoDataFoundError, DataSourceError, ValueError) as e: logger.warning( f"Caught known error fetching basic info for {code}: {type(e).__name__}") raise e except Exception as e: logger.exception( f"Unexpected error fetching basic info for {code}: {e}") raise DataSourceError( f"Unexpected error fetching basic info for {code}: {e}")
• src/data_source_interface.py:67-86 (schema)

  The abstract method definition in the FinancialDataSource interface, specifying the expected input/output schema for get_stock_basic_info (note: tool adds optional fields and format params).
  @abstractmethod def get_stock_basic_info(self, code: str) -> pd.DataFrame: """ Fetches basic information for a given stock code. Args: code: The stock code (e.g., 'sh.600000', 'sz.000001'). Returns: A pandas DataFrame containing the basic stock information. The structure and columns depend on the underlying data source. Typically contains info like name, industry, listing date, etc. 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 the input code is invalid. """ pass