Cryo MCP Server

Instructions

List all available parquet files that can be queried with SQL

USAGE NOTES:
- This function lists parquet files that have already been downloaded
- Each file can be queried using read_parquet('/path/to/file.parquet') in your SQL
- For each file, this returns the file path, dataset type, and other metadata
- Use these file paths in your SQL queries with query_sql()

Returns:
    List of available files and their metadata

Implementation Reference

• cryo_mcp/server.py:694-710 (handler)

  The MCP tool handler for 'list_available_sql_tables', decorated with @mcp.tool() for registration, which delegates to the core helper function in sql.py.

  @mcp.tool()
  def list_available_sql_tables() -> List[Dict[str, Any]]:
      """
      List all available parquet files that can be queried with SQL
      
      USAGE NOTES:
      - This function lists parquet files that have already been downloaded
      - Each file can be queried using read_parquet('/path/to/file.parquet') in your SQL
      - For each file, this returns the file path, dataset type, and other metadata
      - Use these file paths in your SQL queries with query_sql()
      
      Returns:
          List of available files and their metadata
      """
      from cryo_mcp.sql import list_available_tables
      return list_available_tables()

• cryo_mcp/sql.py:34-73 (helper)

  Core helper function that implements the logic: scans the data directory for parquet files, extracts metadata like name, path, size, block range, and returns list of dicts.

  def list_available_tables() -> List[Dict[str, Any]]:
      """List all available tables from downloaded data files."""
      data_dir = get_data_directory()
      
      # Find all parquet files in the data directory (including the latest subdirectory)
      parquet_files = list(data_dir.glob("**/*.parquet"))
      
      tables = []
      for file_path in parquet_files:
          # Extract dataset name from filename
          name = file_path.stem.split("__")[0]
          if "__" in file_path.stem:
              name = file_path.stem.split("__")[0]
          else:
              # Try to extract from other naming patterns
              name_match = re.match(r'([a-z_]+)_', file_path.stem)
              if name_match:
                  name = name_match.group(1)
              else:
                  name = file_path.stem
          
          # Get file stats
          stats = file_path.stat()
          
          # Try to extract block range from filename
          block_range = ""
          blocks_match = re.search(r'blocks__(\d+)_to_(\d+)', str(file_path))
          if blocks_match:
              block_range = f"{blocks_match.group(1)}:{blocks_match.group(2)}"
          
          tables.append({
              "name": name,
              "path": str(file_path),
              "size_bytes": stats.st_size,
              "modified": stats.st_mtime,
              "block_range": block_range,
              "is_latest": "latest" in str(file_path)
          })
      
      return tables