generate_table_insights
Extract AI-driven insights from table data, enabling natural language-based analysis to support core database operations and enhance data exploration.
Instructions
Generate AI-powered insights about a table's data
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| sample_limit | No | ||
| table_name | Yes |
Implementation Reference
- src/main.py:372-399 (handler)Main handler function decorated with @mcp.tool(), implementing the core logic: fetches table schema and sample data from Snowflake, then generates AI insights using OpenAI.async def generate_table_insights(table_name: str, sample_limit: int = 20, ctx: Context = None) -> str: """Generate AI-powered insights about a table's data""" await ctx.info(f"Generating insights for table: {table_name}") try: snowflake = await get_snowflake_client() # Get table schema columns = await snowflake.describe_table(table_name) # Get sample data sample_result = await snowflake.get_table_sample(table_name, sample_limit) if not sample_result.success: await ctx.error(f"Failed to get sample data: {sample_result.error}") return f"Failed to get sample data: {sample_result.error}" # Generate insights openai = await get_openai_client() insights = await openai.generate_data_insights(table_name, columns, sample_result.data) await ctx.info(f"Generated insights for table {table_name}") return insights except Exception as e: logger.error(f"Error generating table insights: {str(e)}") await ctx.error(f"Failed to generate table insights: {str(e)}") raise
- src/openai_client.py:212-272 (helper)Helper method in OpenAIClient that generates AI-powered insights from table schema and sample data using GPT.async def generate_data_insights( self, table_name: str, columns: List[ColumnInfo], sample_data: List[Dict[str, Any]] ) -> str: """Generate insights about a table's data""" # Build column information column_info = [] for col in columns: column_info.append({ 'name': col.column_name, 'type': col.data_type, 'nullable': col.is_nullable, 'comment': col.comment }) system_prompt = """ You are a data analyst. Analyze the provided table schema and sample data to generate insights. Provide: - Data quality assessment - Patterns and trends in the data - Potential use cases for this data - Recommendations for analysis - Any data quality issues you notice Be specific and actionable in your insights. """ user_prompt = f""" Table: {table_name} Column Schema: {json.dumps(column_info, indent=2)} Sample Data: {json.dumps(sample_data, indent=2, default=str)} Please provide insights about this data. """ try: response = await self.client.chat.completions.create( model=self.model, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt} ], temperature=0.4, max_tokens=1000 ) insights = response.choices[0].message.content.strip() logger.info(f"Generated insights for table {table_name}") return insights except Exception as e: logger.error(f"Error generating insights: {str(e)}") raise Exception(f"Failed to generate insights: {str(e)}")
- src/snowflake_client.py:160-182 (helper)SnowflakeClient method to describe table schema (columns), called by the handler.async def describe_table(self, table_name: str, database: Optional[str] = None, schema: Optional[str] = None) -> List[ColumnInfo]: """Get detailed information about a table's columns""" full_table_name = table_name if database and schema: full_table_name = f"{database}.{schema}.{table_name}" elif schema: full_table_name = f"{schema}.{table_name}" result = await self.execute_query(f"DESCRIBE TABLE {full_table_name}") columns = [] for row in result.data: if result.success: columns.append(ColumnInfo( column_name=row.get('name', ''), data_type=row.get('type', ''), is_nullable=row.get('null?', 'Y') == 'Y', default_value=row.get('default'), comment=row.get('comment') )) return columns
- src/snowflake_client.py:183-187 (helper)SnowflakeClient method to fetch sample data from table, called by the handler.async def get_table_sample(self, table_name: str, limit: int = 10) -> QueryResult: """Get a sample of data from a table""" query = f"SELECT * FROM {table_name} LIMIT {limit}" return await self.execute_query(query)