get_product_info
Retrieve detailed product and business composition data for Chinese companies by stock symbol to analyze corporate operations and market positioning.
Instructions
获取公司主要产品/业务构成
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| symbol | Yes | 股票代码 (例如: '000001') | |
| output_format | No | 输出数据格式: json, csv, xml, excel, markdown, html。默认: markdown | markdown |
Implementation Reference
- src/china_stock_mcp/server.py:663-690 (handler)The main handler function for the 'get_product_info' tool. It is decorated with @mcp.tool for registration. Fetches company product/business info using a fallback mechanism between 'ths' (ak.stock_zyjs_ths) and 'eastmoney' (ak.stock_zygc_em) data sources, handles empty DataFrames, and formats the output using _format_dataframe_output.@mcp.tool(name="get_product_info", description="获取公司主要产品/业务构成") def get_product_info( symbol: Annotated[str, Field(description="股票代码 (例如: '000001')")], output_format: Annotated[ Literal["json", "csv", "xml", "excel", "markdown", "html"], Field(description="输出数据格式: json, csv, xml, excel, markdown, html。默认: markdown"), ] = "markdown" ) -> str: """获取产品情况""" def get_product_info_fetcher( source:str,**kwargs: Any ) -> pd.DataFrame: if source == "ths": df = ak.stock_zyjs_ths(symbol) # elif data_source == "eastmoney": else: df = ak.stock_zygc_em(symbol) return df df = _fetch_data_with_fallback( fetch_func=get_product_info_fetcher, primary_source="ths", fallback_sources=["eastmoney"], symbol=symbol, ) if df.empty: df = pd.DataFrame() return _format_dataframe_output(df, output_format)
- Inner helper function that fetches product data based on the source: 'ths' uses ak.stock_zyjs_ths, default (eastmoney) uses ak.stock_zygc_em.def get_product_info_fetcher( source:str,**kwargs: Any ) -> pd.DataFrame: if source == "ths": df = ak.stock_zyjs_ths(symbol) # elif data_source == "eastmoney": else: df = ak.stock_zygc_em(symbol) return df
- src/china_stock_mcp/server.py:15-71 (helper)Shared utility helper for fetching data with primary source and fallbacks, used by get_product_info.def _fetch_data_with_fallback( fetch_func: Callable[..., pd.DataFrame], primary_source: str, fallback_sources: List[str], **kwargs: Any, ) -> pd.DataFrame: """ 通用的数据源故障切换辅助函数。 按优先级尝试数据源，直到获取到有效数据或所有数据源都失败。 Args: fetch_func: 实际调用 akshare 或 akshare_one 获取数据的函数。 这个函数应该接受 'source' 参数，或者在内部处理 source 的映射。 primary_source: 用户指定的首选数据源。 fallback_sources: 备用数据源列表，按优先级排序。 **kwargs: 传递给 fetch_func 的其他参数。 Returns: pd.DataFrame: 获取到的数据。 Raises: RuntimeError: 如果所有数据源都未能获取到有效数据。 """ if primary_source is None: return fetch_func(**kwargs) data_source_priority = [primary_source] + fallback_sources # 移除重复项并保持顺序 seen = set() unique_data_source_priority = [] for x in data_source_priority: if x not in seen: unique_data_source_priority.append(x) seen.add(x) df = None errors = [] for current_source in unique_data_source_priority: try: # 假设 fetch_func 能够接受 source 参数 # 或者 fetch_func 内部根据 kwargs 中的 source 参数进行逻辑判断 temp_df = fetch_func(source=current_source, **kwargs) if temp_df is not None and not temp_df.empty: print(f"成功从数据源 '{current_source}' 获取数据。") df = temp_df break else: errors.append(f"数据源 '{current_source}' 返回空数据。") except Exception as e: errors.append(f"从数据源 '{current_source}' 获取数据失败: {str(e)}") if df is None or df.empty: raise RuntimeError( f"所有数据源都未能获取到有效数据。详细错误: {'; '.join(errors)}" ) return df
- src/china_stock_mcp/server.py:89-118 (helper)Shared utility to format DataFrame output in various formats (json, csv, etc.), used by the handler.def _format_dataframe_output( df: pd.DataFrame, output_format: Literal["json", "csv", "xml", "excel", "markdown", "html"], ) -> str: """ 根据指定的格式格式化 DataFrame 输出。 """ if df.empty: return json.dumps([]) if output_format == "json": return df.to_json(orient="records", force_ascii=False) elif output_format == "csv": return df.to_csv(index=False) elif output_format == "xml": return df.to_xml(index=False) elif output_format == "excel": # 使用 BytesIO 将 Excel 写入内存 output = io.BytesIO() df.to_excel(output, index=False, engine='openpyxl') # 返回 base64 编码的二进制数据，或者直接返回字节流 # 为了兼容性，这里尝试返回 utf-8 编码的字符串，但对于二进制文件，通常直接传输字节流更合适 return output.getvalue().decode("utf-8", errors="ignore") elif output_format == "markdown": return df.to_markdown(index=False) elif output_format == "html": return df.to_html(index=False) else: return df.to_json(orient="records", force_ascii=False)