list_tables
Retrieve a comprehensive list of all tables within a specified database or schema using the DataPilot MCP Server. Simplify database exploration and organization with this essential tool.
Instructions
List all tables in a database/schema
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| database | No | ||
| schema | No |
Implementation Reference
- src/main.py:139-174 (handler)Primary MCP tool handler for 'list_tables'. Decorated with @mcp.tool(), handles parameters, calls SnowflakeClient.list_tables(), formats results as list of dicts, and provides context messages.@mcp.tool() async def list_tables(database: Optional[str] = None, schema: Optional[str] = None, ctx: Context = None) -> List[Dict[str, Any]]: """List all tables in a database/schema""" context_msg = f"Retrieving tables" if database: context_msg += f" from database: {database}" if schema: context_msg += f", schema: {schema}" await ctx.info(context_msg) try: client = await get_snowflake_client() tables = await client.list_tables(database, schema) # Convert to dict for JSON serialization result = [] for table in tables: result.append({ "table_name": table.table_name, "schema_name": table.schema_name, "database_name": table.database_name, "table_type": table.table_type, "row_count": table.row_count, "bytes": table.bytes, "comment": table.comment }) await ctx.info(f"Found {len(result)} tables") return result except Exception as e: logger.error(f"Error listing tables: {str(e)}") await ctx.error(f"Failed to list tables: {str(e)}") return []
- src/snowflake_client.py:135-159 (helper)Supporting method in SnowflakeClient class that constructs and executes the SHOW TABLES query, parses results into TableInfo objects.async def list_tables(self, database: Optional[str] = None, schema: Optional[str] = None) -> List[TableInfo]: """List all tables in a database/schema""" query = "SHOW TABLES" if database and schema: query += f" IN SCHEMA {database}.{schema}" elif database: query += f" IN DATABASE {database}" result = await self.execute_query(query) tables = [] for row in result.data: if result.success: tables.append(TableInfo( table_name=row.get('name', ''), schema_name=row.get('schema_name', ''), database_name=row.get('database_name', ''), table_type=row.get('kind', ''), row_count=row.get('rows'), bytes=row.get('bytes'), comment=row.get('comment') )) return tables
- src/main.py:139-139 (registration)The @mcp.tool() decorator registers the list_tables function as an MCP tool.@mcp.tool()
- src/main.py:428-448 (helper)MCP resource that provides table listing functionality similar to the tool, used for resource-based access.@mcp.resource("snowflake://tables/{database}/{schema}") async def get_tables_resource(database: str, schema: str) -> List[Dict[str, Any]]: """Resource to get list of tables in a database/schema""" try: client = await get_snowflake_client() tables = await client.list_tables(database, schema) return [ { "table_name": table.table_name, "schema_name": table.schema_name, "database_name": table.database_name, "table_type": table.table_type, "row_count": table.row_count, "bytes": table.bytes, "comment": table.comment } for table in tables ] except Exception as e: logger.error(f"Error getting tables resource: {str(e)}") return []