Cryo MCP Server

Instructions

Implementation Reference

• cryo_mcp/server.py:322-477 (handler)

  The primary handler function for the 'lookup_dataset' tool. It is decorated with @mcp.tool() for registration in the MCP server. The function fetches dataset information using get_dataset_info, retrieves schema via cryo's --dry-run, and optionally generates sample data by querying a small block range.
  @mcp.tool() def lookup_dataset( name: str, sample_start_block: Optional[int] = None, sample_end_block: Optional[int] = None, use_latest_sample: bool = False, sample_blocks_from_latest: Optional[int] = None ) -> Dict[str, Any]: """ Look up a specific dataset and return detailed information about it. IMPORTANT: Always use this function before querying a new dataset to understand its required parameters and schema. The returned information includes: 1. Required parameters for the dataset (IMPORTANT for datasets like 'balances' that need an address) 2. Schema details showing available columns and data types 3. Example queries for the dataset When the dataset requires specific parameters like 'address' (for 'balances'), ALWAYS use the 'contract' parameter in query_dataset() to pass these values. Example: For 'balances' dataset, lookup_dataset('balances') will show it requires an 'address' parameter. You should then query it using: query_dataset('balances', blocks='1000:1010', contract='0x1234...') Args: name: The name of the dataset to look up sample_start_block: Optional start block for sample data (integer) sample_end_block: Optional end block for sample data (integer) use_latest_sample: If True, use the latest block for sample data sample_blocks_from_latest: Number of blocks before the latest to include in sample Returns: Detailed information about the dataset including schema and available fields """ # Get basic dataset info info = get_dataset_info(name) # Ensure we have the RPC URL rpc_url = os.environ.get("ETH_RPC_URL", DEFAULT_RPC_URL) # Get schema information by running the dataset with --dry-run schema_result = subprocess.run( ["cryo", name, "--dry-run", "-r", rpc_url], capture_output=True, text=True ) if schema_result.returncode == 0: info["schema"] = schema_result.stdout else: info["schema_error"] = schema_result.stderr # Try to get a sample of the dataset (first 5 records) try: data_dir = Path(os.environ.get("CRYO_DATA_DIR", DEFAULT_DATA_DIR)) # Determine block range for sample (priority: latest > specified blocks > default) if use_latest_sample or sample_blocks_from_latest is not None: # Get the latest block number latest_block = get_latest_block_number() if latest_block is None: info["sample_error"] = "Failed to get the latest block number from the RPC endpoint" return info if sample_blocks_from_latest is not None: # Use a range of blocks from latest-n to latest block_range = f"{latest_block - sample_blocks_from_latest}:{latest_block+1}" else: # Just the latest 5 blocks block_range = f"{latest_block-4}:{latest_block+1}" info["sample_block_range"] = block_range # Use the latest directory for latest block samples sample_dir = data_dir / "latest" sample_dir.mkdir(parents=True, exist_ok=True) # Clean up the latest directory before new query print("Cleaning latest directory for current sample") existing_files = list(sample_dir.glob(f"*{name}*.*")) for file in existing_files: try: file.unlink() print(f"Removed existing sample file: {file}") except Exception as e: print(f"Warning: Could not remove sample file {file}: {e}") else: # For historical blocks, get the start block and end block if sample_start_block is not None: if sample_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"{sample_start_block}:{sample_end_block+1}" else: # Use start block and get 5 blocks block_range = f"{sample_start_block}:{sample_start_block+5}" else: # Default to a known good block range block_range = "1000:1005" # For historical samples, use the main data directory sample_dir = data_dir sample_dir.mkdir(parents=True, exist_ok=True) # Use the block range for the sample sample_cmd = [ "cryo", name, "-b", block_range, "-r", rpc_url, "--json", "-o", str(sample_dir) ] print(f"Running sample command: {' '.join(sample_cmd)}") sample_result = subprocess.run( sample_cmd, capture_output=True, text=True, timeout=30 # Add timeout to prevent hanging ) if sample_result.returncode == 0: # Try to find the report file which contains info about generated files report_dir = sample_dir / ".cryo" / "reports" if report_dir.exists(): # Get the most recent report file 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}") info["sample_files"] = completed_files return info # Fallback to glob search if report file not found output_files = list(sample_dir.glob(f"*{name}*.json")) print(f"Output files found via glob: {output_files}") if output_files: # Convert Path objects to strings for JSON serialization file_paths = [str(file_path) for file_path in output_files] info["sample_files"] = file_paths else: info["sample_error"] = "No output files generated" else: info["sample_error"] = sample_result.stderr info["sample_stdout"] = sample_result.stdout # Include stdout for debugging except (subprocess.TimeoutExpired, Exception) as e: info["sample_error"] = str(e) return info
• cryo_mcp/server.py:274-320 (helper)

  Helper resource function called by lookup_dataset to get basic dataset information, description, and example queries via cryo's help command.
  @mcp.resource("dataset://{name}") def get_dataset_info(name: str) -> Dict[str, Any]: """Get information about a specific dataset""" # Ensure we have the RPC URL rpc_url = os.environ.get("ETH_RPC_URL", DEFAULT_RPC_URL) result = subprocess.run( ["cryo", "help", name, "-r", rpc_url], capture_output=True, text=True ) # Get the latest block number for examples latest_block = get_latest_block_number() latest_example = "" if latest_block: latest_example = f"query_dataset('{name}', blocks_from_latest=10) # Gets latest-10 to latest blocks" # Add special examples for datasets requiring address parameters address_example = "" if "address" in result.stdout.lower() and "required parameters: address" in result.stdout.lower(): address_example = f"query_dataset('{name}', blocks='1000:1010', contract='0x123...') # Use contract parameter for address" return { "name": name, "description": result.stdout, "example_queries": [ f"query_dataset('{name}', blocks='1000:1010')", f"query_dataset('{name}', start_block=1000, end_block=1009)", f"query_dataset('{name}', use_latest=True) # Gets just the latest block", latest_example, address_example ] if address_example else [ f"query_dataset('{name}', blocks='1000:1010')", f"query_dataset('{name}', start_block=1000, end_block=1009)", f"query_dataset('{name}', use_latest=True) # Gets just the latest block", latest_example ], "notes": [ "Block ranges are inclusive for start_block and end_block when using integer parameters.", "Use 'use_latest=True' to query only the latest block.", "Use 'blocks_from_latest=N' to query the latest N blocks.", "IMPORTANT: For datasets requiring an 'address' parameter (like 'balances'), use the 'contract' parameter.", "Always check the required parameters in the dataset description and use lookup_dataset() first." ] }
• cryo_mcp/server.py:322-322 (registration)

  The @mcp.tool() decorator registers the lookup_dataset function as an MCP tool.
  @mcp.tool()