Panther MCP Server

Instructions

Implementation Reference

• src/mcp_panther/panther_mcp_core/tools/data_lake.py:627-703 (handler)

  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)}"}