Cryo MCP Server

Instructions

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