A-Share MCP Server

Instructions

Implementation Reference

• src/tools/financial_reports.py:99-123 (handler)

  The primary MCP tool handler for 'get_fina_indicator'. Decorated with @app.tool() for automatic registration. It invokes the use case with standardized error handling, logging context, and output formatting.
  @app.tool() def get_fina_indicator(code: str, start_date: str, end_date: str, limit: int = 250, format: str = "markdown") -> str: """ Aggregated financial indicators from 6 Baostock APIs into one convenient query. **Data is returned by QUARTER** (Q1, Q2, Q3, Q4) based on financial report dates. Input date range determines which quarters to fetch. Combines data from: - 盈利能力 (Profitability): roeAvg, npMargin, gpMargin, epsTTM - 营运能力 (Operation): NRTurnRatio, INVTurnRatio, CATurnRatio - 成长能力 (Growth): YOYNI, YOYEquity, YOYAsset - 偿债能力 (Solvency): currentRatio, quickRatio, liabilityToAsset - 现金流量 (Cash Flow): CFOToOR, CFOToNP, CAToAsset - 杜邦分析 (DuPont): dupontROE, dupontAssetTurn, dupontPnitoni Output columns include prefixes: profit_*, operation_*, growth_*, balance_*, cashflow_*, dupont_* to distinguish data sources. """ return run_tool_with_handling( lambda: fetch_fina_indicator( active_data_source, code=code, start_date=start_date, end_date=end_date, limit=limit, format=format ), context=f"get_fina_indicator:{code}:{start_date}-{end_date}", )
• mcp_server.py:52-52 (registration)

  Invocation of the registration function that adds the get_fina_indicator tool (and other financial report tools) to the FastMCP app instance.
  register_financial_report_tools(app, active_data_source)
• src/data_source_interface.py:166-184 (schema)

  Interface definition for get_fina_indicator method, providing input/output schema and documentation used throughout the tool chain.
  @abstractmethod def get_fina_indicator(self, code: str, start_date: str, end_date: str) -> pd.DataFrame: """ Fetches financial indicators (ROE, gross margin, net margin, etc.) within a date range. 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. Returns: A pandas DataFrame containing financial indicators such as: - roe, roe_yearly (Return on Equity) - netprofit_margin, grossprofit_margin (Profitability ratios) - expense_ratio, netprofit_ratio - current_ratio, quick_ratio (Liquidity ratios) - etc. """ pass
• src/use_cases/financial_reports.py:84-89 (helper)

  Use case layer function that orchestrates data fetching from the data source and applies output formatting and validation.
  def fetch_fina_indicator(data_source: FinancialDataSource, *, code: str, start_date: str, end_date: str, limit: int, format: str) -> str: """Fetch financial indicators (ROE, gross margin, net margin, etc.) within a date range.""" validate_output_format(format) df = data_source.get_fina_indicator(code=code, start_date=start_date, end_date=end_date) meta = {"code": code, "start_date": start_date, "end_date": end_date, "dataset": "Financial Indicators"} return format_table_output(df, format=format, max_rows=limit, meta=meta)
• src/baostock_data_source.py:687-803 (helper)

  Core implementation in BaostockDataSource that aggregates data from 6 different Baostock APIs (profit, operation, growth, balance, cashflow, dupont) into a single prefixed DataFrame for each quarter in the date range.
  def get_fina_indicator(self, code: str, start_date: str, end_date: str) -> pd.DataFrame: """ Fetches comprehensive financial indicators by aggregating multiple Baostock APIs. Aggregates data from: - Profitability (盈利能力) - Operation Capability (营运能力) - Growth Capability (成长能力) - Balance Sheet/Solvency (偿债能力) - Cash Flow (现金流量) - DuPont Analysis (杜邦分析) """ logger.info(f"Fetching aggregated financial indicators for {code} ({start_date} to {end_date})") # 解析日期范围，获取年份列表 from datetime import datetime try: start = datetime.strptime(start_date, "%Y-%m-%d") end = datetime.strptime(end_date, "%Y-%m-%d") except ValueError: raise ValueError(f"Invalid date format. Expected YYYY-MM-DD, got {start_date} to {end_date}") years = set(str(y) for y in range(start.year, end.year + 1)) all_results = [] try: with baostock_login_context(): for year in years: for quarter in [1, 2, 3, 4]: # 检查季度是否在日期范围内 quarter_start_month = (quarter - 1) * 3 quarter_start = datetime(int(year), quarter_start_month + 1, 1) if quarter_start > end: continue record = {"code": code, "year": year, "quarter": quarter} # 1. 盈利能力 try: rs = bs.query_profit_data(code=code, year=year, quarter=quarter) if rs.error_code == '0' and rs.next(): row = rs.get_row_data() for i, field in enumerate(rs.fields): record[f"profit_{field}"] = row[i] if i < len(row) else None except Exception as e: logger.debug(f"Failed to fetch profit data for {code} {year}Q{quarter}: {e}") # 2. 营运能力 try: rs = bs.query_operation_data(code=code, year=year, quarter=quarter) if rs.error_code == '0' and rs.next(): row = rs.get_row_data() for i, field in enumerate(rs.fields): record[f"operation_{field}"] = row[i] if i < len(row) else None except Exception as e: logger.debug(f"Failed to fetch operation data for {code} {year}Q{quarter}: {e}") # 3. 成长能力 try: rs = bs.query_growth_data(code=code, year=year, quarter=quarter) if rs.error_code == '0' and rs.next(): row = rs.get_row_data() for i, field in enumerate(rs.fields): record[f"growth_{field}"] = row[i] if i < len(row) else None except Exception as e: logger.debug(f"Failed to fetch growth data for {code} {year}Q{quarter}: {e}") # 4. 偿债能力 try: rs = bs.query_balance_data(code=code, year=year, quarter=quarter) if rs.error_code == '0' and rs.next(): row = rs.get_row_data() for i, field in enumerate(rs.fields): record[f"balance_{field}"] = row[i] if i < len(row) else None except Exception as e: logger.debug(f"Failed to fetch balance data for {code} {year}Q{quarter}: {e}") # 5. 现金流量 try: rs = bs.query_cash_flow_data(code=code, year=year, quarter=quarter) if rs.error_code == '0' and rs.next(): row = rs.get_row_data() for i, field in enumerate(rs.fields): record[f"cashflow_{field}"] = row[i] if i < len(row) else None except Exception as e: logger.debug(f"Failed to fetch cash flow data for {code} {year}Q{quarter}: {e}") # 6. 杜邦分析 try: rs = bs.query_dupont_data(code=code, year=year, quarter=quarter) if rs.error_code == '0' and rs.next(): row = rs.get_row_data() for i, field in enumerate(rs.fields): record[f"dupont_{field}"] = row[i] if i < len(row) else None except Exception as e: logger.debug(f"Failed to fetch dupont data for {code} {year}Q{quarter}: {e}") # 只有当有数据时才添加记录 if len(record) > 3: # code + year + quarter + at least one data field all_results.append(record) if not all_results: raise NoDataFoundError( f"No financial indicator data found for {code} in range {start_date}-{end_date}") result_df = pd.DataFrame(all_results) logger.info(f"Retrieved {len(result_df)} aggregated financial indicator records for {code}.") return result_df except (LoginError, NoDataFoundError, DataSourceError, ValueError) as e: logger.warning(f"Known error fetching financial indicators for {code}: {type(e).__name__}") raise e except Exception as e: logger.exception(f"Unexpected error fetching financial indicators for {code}: {e}") raise DataSourceError(f"Unexpected error fetching financial indicators for {code}: {e}")