Cryo MCP Server

Instructions

Download blockchain data and return the file paths where the data is stored.

IMPORTANT WORKFLOW NOTE: When running SQL queries, use this function first to download
data, then use the returned file paths with query_sql() to execute SQL on those files.

Example workflow for SQL:
1. First download data: result = query_dataset('transactions', blocks='1000:1010', output_format='parquet')
2. Get file paths: files = result.get('files', [])
3. Run SQL query: query_sql("SELECT * FROM read_parquet('/path/to/file.parquet')", files=files)

DATASET-SPECIFIC PARAMETERS:
For datasets that require specific address parameters (like 'balances', 'erc20_transfers', etc.),
ALWAYS use the 'contract' parameter to pass ANY Ethereum address. For example:

- For 'balances' dataset: Use contract parameter for the address you want balances for
  query_dataset('balances', blocks='1000:1010', contract='0x123...')

- For 'logs' or 'erc20_transfers': Use contract parameter for contract address
  query_dataset('logs', blocks='1000:1010', contract='0x123...')

To check what parameters a dataset requires, always use lookup_dataset() first:
lookup_dataset('balances')  # Will show required parameters

Args:
    dataset: The name of the dataset to query (e.g., 'logs', 'transactions', 'balances')
    blocks: Block range specification as a string (e.g., '1000:1010')
    start_block: Start block number as integer (alternative to blocks)
    end_block: End block number as integer (alternative to blocks)
    use_latest: If True, query the latest block
    blocks_from_latest: Number of blocks before the latest to include (e.g., 10 = latest-10 to latest)
    contract: Contract address to filter by - IMPORTANT: Use this parameter for ALL address-based filtering
      regardless of the parameter name in the native cryo command (address, contract, etc.)
    output_format: Output format (json, csv, parquet) - use 'parquet' for SQL queries
    include_columns: Columns to include alongside the defaults
    exclude_columns: Columns to exclude from the defaults

Returns:
    Dictionary containing file paths where the downloaded data is stored

Implementation Reference

• cryo_mcp/server.py:80-272 (handler)

  The query_dataset tool handler function, registered via @mcp.tool() decorator. Executes cryo CLI to download blockchain dataset data for specified parameters and returns file paths to the generated files.

  @mcp.tool()
  def query_dataset(
      dataset: str,
      blocks: Optional[str] = None,
      start_block: Optional[int] = None,
      end_block: Optional[int] = None,
      use_latest: bool = False,
      blocks_from_latest: Optional[int] = None,
      contract: Optional[str] = None,
      output_format: str = "json",
      include_columns: Optional[List[str]] = None,
      exclude_columns: Optional[List[str]] = None
  ) -> Dict[str, Any]:
      """
      Download blockchain data and return the file paths where the data is stored.
      
      IMPORTANT WORKFLOW NOTE: When running SQL queries, use this function first to download
      data, then use the returned file paths with query_sql() to execute SQL on those files.
      
      Example workflow for SQL:
      1. First download data: result = query_dataset('transactions', blocks='1000:1010', output_format='parquet')
      2. Get file paths: files = result.get('files', [])
      3. Run SQL query: query_sql("SELECT * FROM read_parquet('/path/to/file.parquet')", files=files)
  
      DATASET-SPECIFIC PARAMETERS:
      For datasets that require specific address parameters (like 'balances', 'erc20_transfers', etc.),
      ALWAYS use the 'contract' parameter to pass ANY Ethereum address. For example:
      
      - For 'balances' dataset: Use contract parameter for the address you want balances for
        query_dataset('balances', blocks='1000:1010', contract='0x123...')
      
      - For 'logs' or 'erc20_transfers': Use contract parameter for contract address
        query_dataset('logs', blocks='1000:1010', contract='0x123...')
      
      To check what parameters a dataset requires, always use lookup_dataset() first:
      lookup_dataset('balances')  # Will show required parameters
  
      Args:
          dataset: The name of the dataset to query (e.g., 'logs', 'transactions', 'balances')
          blocks: Block range specification as a string (e.g., '1000:1010')
          start_block: Start block number as integer (alternative to blocks)
          end_block: End block number as integer (alternative to blocks)
          use_latest: If True, query the latest block
          blocks_from_latest: Number of blocks before the latest to include (e.g., 10 = latest-10 to latest)
          contract: Contract address to filter by - IMPORTANT: Use this parameter for ALL address-based filtering
            regardless of the parameter name in the native cryo command (address, contract, etc.)
          output_format: Output format (json, csv, parquet) - use 'parquet' for SQL queries
          include_columns: Columns to include alongside the defaults
          exclude_columns: Columns to exclude from the defaults
  
      Returns:
          Dictionary containing file paths where the downloaded data is stored
      """
      # Ensure we have the RPC URL
      rpc_url = os.environ.get("ETH_RPC_URL", DEFAULT_RPC_URL)
      
      # Build the cryo command
      cmd = ["cryo", dataset, "-r", rpc_url]
  
      # Handle block range (priority: blocks > use_latest > start/end_block > default)
      if blocks:
          # Use specified block range string directly
          cmd.extend(["-b", blocks])
      elif use_latest or blocks_from_latest is not None:
          # Get the latest block number
          latest_block = get_latest_block_number()
          
          if latest_block is None:
              return {"error": "Failed to get the latest block number from the RPC endpoint"}
          
          if blocks_from_latest is not None:
              # Use a range of blocks up to the latest
              start = latest_block - blocks_from_latest
              block_range = f"{start}:{latest_block+1}"  # +1 to make it inclusive
          else:
              # Just the latest block
              block_range = f"{latest_block}:{latest_block+1}"  # +1 to make it inclusive
          
          print(f"Using latest block range: {block_range}")
          cmd.extend(["-b", block_range])
      elif start_block is not None:
          # Convert integer block numbers to string range
          if end_block is not None:
              # Note: cryo uses [start:end) range (inclusive start, exclusive end)
              # Add 1 to end_block to include it in the range
              block_range = f"{start_block}:{end_block+1}"
          else:
              # If only start_block is provided, get 10 blocks starting from there
              block_range = f"{start_block}:{start_block+10}"
          
          print(f"Using block range: {block_range}")
          cmd.extend(["-b", block_range])
      else:
          # Default to a reasonable block range if none specified
          cmd.extend(["-b", "1000:1010"])
  
      # Handle dataset-specific address parameters
      # For all address-based filters, we use the contract parameter
      # but map it to the correct flag based on the dataset
      if contract:
          # Check if this is a dataset that requires a different parameter name
          if dataset == 'balances':
              # For balances dataset, contract parameter maps to --address
              cmd.extend(["--address", contract])
          else:
              # For other datasets like logs, transactions, etc. use --contract
              cmd.extend(["--contract", contract])
  
      if output_format == "json":
          cmd.append("--json")
      elif output_format == "csv":
          cmd.append("--csv")
  
      if include_columns:
          cmd.append("--include-columns")
          cmd.extend(include_columns)
  
      if exclude_columns:
          cmd.append("--exclude-columns")
          cmd.extend(exclude_columns)
  
      # Get the base data directory
      data_dir = Path(os.environ.get("CRYO_DATA_DIR", DEFAULT_DATA_DIR))
      
      # Choose output directory based on whether we're querying latest blocks
      if use_latest or blocks_from_latest is not None:
          output_dir = data_dir / "latest"
          output_dir.mkdir(parents=True, exist_ok=True)
          
          # Clean up the latest directory before new query
          print("Cleaning latest directory for current block query")
          existing_files = list(output_dir.glob(f"*{dataset}*.*"))
          for file in existing_files:
              try:
                  file.unlink()
                  print(f"Removed existing file: {file}")
              except Exception as e:
                  print(f"Warning: Could not remove file {file}: {e}")
      else:
          # For historical queries, use the main data directory
          output_dir = data_dir
          output_dir.mkdir(parents=True, exist_ok=True)
  
      cmd.extend(["-o", str(output_dir)])
  
      # Print the command for debugging
      print(f"Running query command: {' '.join(cmd)}")
      
      # Execute the command
      result = subprocess.run(cmd, capture_output=True, text=True)
  
      if result.returncode != 0:
          return {
              "error": result.stderr,
              "stdout": result.stdout,
              "command": " ".join(cmd)
          }
  
      # Try to find the report file which contains info about generated files
      report_dir = output_dir / ".cryo" / "reports"
      if report_dir.exists():
          # Get the most recent report file (should be the one we just created)
          report_files = sorted(report_dir.glob("*.json"), key=lambda x: x.stat().st_mtime, reverse=True)
          if report_files:
              with open(report_files[0], 'r') as f:
                  report_data = json.load(f)
                  # Get the list of completed files from the report
                  if "results" in report_data and "completed_paths" in report_data["results"]:
                      completed_files = report_data["results"]["completed_paths"]
                      print(f"Found {len(completed_files)} files in Cryo report: {completed_files}")
                      
                      # Return the list of files and their count
                      return {
                          "files": completed_files,
                          "count": len(completed_files),
                          "format": output_format
                      }
      
      # Fallback to glob search if report file not found or doesn't contain the expected data
      output_files = list(output_dir.glob(f"*{dataset}*.{output_format}"))
      print(f"Output files found via glob: {output_files}")
  
      if not output_files:
          return {"error": "No output files generated", "command": " ".join(cmd)}
  
      # Convert Path objects to strings for JSON serialization
      file_paths = [str(file_path) for file_path in output_files]
      
      return {
          "files": file_paths,
          "count": len(file_paths),
          "format": output_format
      }