DataPilot MCP Server

Instructions

Implementation Reference

• src/main.py:176-202 (handler)

  MCP tool handler for 'describe_table' that fetches column information from Snowflake client and formats it as a list of dictionaries.

  @mcp.tool()
  async def describe_table(table_name: str, database: Optional[str] = None, schema: Optional[str] = None, ctx: Context = None) -> List[Dict[str, Any]]:
      """Get detailed information about a table's columns"""
      await ctx.info(f"Describing table: {table_name}")
      
      try:
          client = await get_snowflake_client()
          columns = await client.describe_table(table_name, database, schema)
          
          # Convert to dict for JSON serialization
          result = []
          for col in columns:
              result.append({
                  "column_name": col.column_name,
                  "data_type": col.data_type,
                  "is_nullable": col.is_nullable,
                  "default_value": col.default_value,
                  "comment": col.comment
              })
          
          await ctx.info(f"Table {table_name} has {len(result)} columns")
          return result
          
      except Exception as e:
          logger.error(f"Error describing table: {str(e)}")
          await ctx.error(f"Failed to describe table: {str(e)}")
          return []

• src/models.py:35-41 (schema)

  Pydantic model defining the structure of column information returned by describe_table.

  class ColumnInfo(BaseModel):
      """Information about a table column"""
      column_name: str
      data_type: str
      is_nullable: bool
      default_value: Optional[str] = None
      comment: Optional[str] = None

• src/snowflake_client.py:160-181 (helper)

  SnowflakeClient method that executes DESCRIBE TABLE query, parses results into ColumnInfo objects.

  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