list_database_tables
Read-only
Retrieve all available tables within a specified Panther Database to understand data structure and log types for security monitoring queries.
Instructions
List all available tables in a Panther Database.
Required: Only use valid database names obtained from list_databases
Returns: Dict containing: - success: Boolean indicating if the query was successful - tables: List of tables, each containing: - name: Table name - description: Table description - log_type: Log type - database: Database name - message: Error message if unsuccessful
Permissions:{'all_of': ['Query Data Lake']}
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| database | Yes | The name of the database to list tables for |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- The core handler function for the 'list_database_tables' tool. It is decorated with @mcp_tool for registration and includes inline schema via Annotated[Field]. Fetches tables from a specified database using GraphQL pagination via LIST_TABLES_QUERY.
@mcp_tool( annotations={ "permissions": all_perms(Permission.DATA_ANALYTICS_READ), "readOnlyHint": True, } ) async def list_database_tables( database: Annotated[ str, Field( description="The name of the database to list tables for", examples=["panther_logs.public"], ), ], ) -> Dict[str, Any]: """List all available tables in a Panther Database. Required: Only use valid database names obtained from list_databases Returns: Dict containing: - success: Boolean indicating if the query was successful - tables: List of tables, each containing: - name: Table name - description: Table description - log_type: Log type - database: Database name - message: Error message if unsuccessful """ logger.info("Fetching available tables") all_tables = [] page_size = 100 try: logger.info(f"Fetching tables for database: {database}") cursor = None while True: # Prepare input variables variables = { "databaseName": database, "pageSize": page_size, "cursor": cursor, } logger.debug(f"Query variables: {variables}") # Execute the query using shared client result = await _execute_query(LIST_TABLES_QUERY, variables) # Get query data result = result.get("dataLakeDatabaseTables", {}) for table in result.get("edges", []): all_tables.append(table["node"]) # Check if there are more pages page_info = result["pageInfo"] if not page_info["hasNextPage"]: break # Update cursor for next page cursor = page_info["endCursor"] # Format the response return { "success": True, "status": "succeeded", "tables": all_tables, "stats": { "table_count": len(all_tables), }, } except Exception as e: logger.error(f"Failed to fetch tables: {str(e)}") return {"success": False, "message": f"Failed to fetch tables: {str(e)}"}