低温 MCP 🧊

Cryo区块链数据提取工具的模型完成协议 (MCP) 服务器。

Cryo MCP 允许您通过实现 MCP 协议的 API 服务器访问 Cryo 强大的区块链数据提取功能，从而可以轻松地从任何兼容 MCP 的客户端查询区块链数据。

对于 LLM 用户：SQL 查询工作流指南

使用此 MCP 服务器对区块链数据运行 SQL 查询时，请遵循以下工作流程：

使用query_dataset下载数据：
result = query_dataset( dataset="blocks", # or "transactions", "logs", etc. blocks="15000000:15001000", # or use blocks_from_latest=100 output_format="parquet" # important: use parquet for SQL ) files = result.get("files", []) # Get the returned file paths
使用get_sql_table_schema探索模式：
# Check what columns are available in the file schema = get_sql_table_schema(files[0]) # Now you can see all columns, data types, and sample data
使用query_sql运行 SQL ：
# Option 1: Simple table reference (DuckDB will match the table name to file) sql_result = query_sql( query="SELECT block_number, timestamp, gas_used FROM blocks", files=files # Pass the files from step 1 ) # Option 2: Using read_parquet() with explicit file path sql_result = query_sql( query=f"SELECT block_number, timestamp, gas_used FROM read_parquet('{files[0]}')", files=files # Pass the files from step 1 )

或者，使用与query_blockchain_sql组合方法：

# Option 1: Simple table reference
result = query_blockchain_sql(
    sql_query="SELECT * FROM blocks",
    dataset="blocks",
    blocks_from_latest=100
)

# Option 2: Using read_parquet()
result = query_blockchain_sql(
    sql_query="SELECT * FROM read_parquet('/path/to/file.parquet')",  # Path doesn't matter
    dataset="blocks",
    blocks_from_latest=100
)

有关完整的工作示例，请参阅examples/sql_workflow_example.py 。

特征

完整 Cryo 数据集访问：通过 API 服务器查询任何 Cryo 数据集
MCP 集成：与 MCP 客户端无缝协作
灵活的查询选项：支持所有主要的Cryo过滤和输出选项
区块范围选项：查询特定区块、最新区块或相对范围
合约过滤：按合约地址过滤数据
最新区块访问：轻松访问最新的以太坊区块数据
多种输出格式：JSON、CSV 和 Parquet 支持
架构信息：获取详细的数据集架构和示例数据
SQL 查询：直接针对下载的区块链数据运行 SQL 查询

安装（可选）

如果您直接使用uvx运行该工具，则不需要这样做。

# install with UV (recommended)
uv tool install cryo-mcp

要求

Python 3.8+
紫外线
Cryo的工作安装
访问以太坊 RPC 端点
DuckDB（用于 SQL 查询功能）

快速入门

与 Claude Code 一起使用

运行claude mcp add以获得交互式提示。
输入uvx作为运行命令。
输入cryo-mcp --rpc-url <ETH_RPC_URL> [--data-dir <DATA_DIR>]作为参数
或者，提供ETH_RPC_URL和CRYO_DATA_DIR作为环境变量。

claude的新实例现在可以访问 cryo，因为它配置为命中您的 RPC 端点并将数据存储在指定的目录中。

可用工具

Cryo MCP 公开了以下 MCP 工具：

`list_datasets()`

返回所有可用 Cryo 数据集的列表。

例子：

client.list_datasets()

`query_dataset()`

使用各种过滤选项查询 Cryo 数据集。

参数：

dataset （str）：要查询的数据集的名称（例如，“blocks”、“transactions”、“logs”）
blocks （str，可选）：块范围规范（例如，'1000：1010'）
start_block (int，可选)：起始块号（块的替代）
end_block （int，可选）：结束块号（块的替代）
use_latest (bool, 可选): 如果为 True，则查询最新区块
blocks_from_latest (int，可选)：从最新到包含的块数
contract （str，可选）：要过滤的合约地址
output_format （str，可选）：输出格式（'json'，'csv'，'parquet'）
include_columns （列表，可选）：与默认值一起包含的列
exclude_columns （列表，可选）：从默认值中排除的列

例子：

# Get transactions from blocks 15M to 15.01M
client.query_dataset('transactions', blocks='15M:15.01M')

# Get logs for a specific contract from the latest 100 blocks
client.query_dataset('logs', blocks_from_latest=100, contract='0x1234...')

# Get just the latest block
client.query_dataset('blocks', use_latest=True)

`lookup_dataset()`

获取有关特定数据集的详细信息，包括模式和示例数据。

参数：

name (str): 要查找的数据集的名称
sample_start_block （int，可选）：样本数据的起始块
sample_end_block （int，可选）：样本数据的结束块
use_latest_sample (bool，可选)：使用最新块作为样本
sample_blocks_from_latest (int，可选)：样本的最新块数

例子：

client.lookup_dataset('logs')

`get_latest_ethereum_block()`

返回有关最新以太坊区块的信息。

例子：

client.get_latest_ethereum_block()

SQL查询工具

Cryo MCP 包含几个用于针对区块链数据运行 SQL 查询的工具：

`query_sql()`

对下载的区块链数据运行 SQL 查询。

参数：

query （str）：要执行的 SQL 查询
files （列表，可选）：要查询的 parquet 文件路径列表。如果为 None，则使用数据目录中的所有文件。
include_schema (bool，可选)：是否在结果中包含架构信息

例子：

# Run against all available files
client.query_sql("SELECT * FROM read_parquet('/path/to/blocks.parquet') LIMIT 10")

# Run against specific files
client.query_sql(
    "SELECT * FROM read_parquet('/path/to/blocks.parquet') LIMIT 10",
    files=['/path/to/blocks.parquet']
)

`query_blockchain_sql()`

使用 SQL 查询区块链数据，自动下载任何所需数据。

参数：

sql_query (str): 要执行的 SQL 查询
dataset （str，可选）：要查询的数据集（例如，“区块”、“交易”）
blocks （str，可选）：块范围规范
start_block （int，可选）：起始块号
end_block （int，可选）：结束块号
use_latest (bool, 可选): 如果为 True，则查询最新区块
blocks_from_latest (int，可选)：要包含的最新块之前块的数量
contract （str，可选）：要过滤的合约地址
force_refresh （bool，可选）：即使存在新数据也强制下载。
include_schema (bool，可选)：在结果中包含架构信息

例子：

# Automatically downloads blocks data if needed, then runs the SQL query
client.query_blockchain_sql(
    sql_query="SELECT block_number, gas_used, timestamp FROM blocks ORDER BY gas_used DESC LIMIT 10",
    dataset="blocks",
    blocks_from_latest=100
)

`list_available_sql_tables()`

列出所有可用 SQL 查询的表。

例子：

client.list_available_sql_tables()

`get_sql_table_schema()`

获取特定 parquet 文件的架构。

参数：

file_path (str): parquet 文件的路径

例子：

client.get_sql_table_schema("/path/to/blocks.parquet")

`get_sql_examples()`

获取不同区块链数据集的示例 SQL 查询。

例子：

client.get_sql_examples()

配置选项

启动 Cryo MCP 服务器时，您可以使用以下命令行选项：

--rpc-url URL ：以太坊 RPC URL（覆盖 ETH_RPC_URL 环境变量）
--data-dir PATH ：存储下载数据的目录（覆盖 CRYO_DATA_DIR 环境变量，默认为 ~/.cryo-mcp/data/）

环境变量

ETH_RPC_URL ：未通过命令行指定时使用的默认以太坊 RPC URL
CRYO_DATA_DIR ：未通过命令行指定时存储下载数据的默认目录

高级用法

针对区块链数据的 SQL 查询

Cryo MCP 允许您对区块链数据运行强大的 SQL 查询，将 SQL 的灵活性与 Cryo 的数据提取功能相结合：

两步 SQL 查询流程

您可以将数据提取和查询分为两个独立的步骤：

# Step 1: Download data and get file paths
download_result = client.query_dataset(
    dataset="transactions",
    blocks_from_latest=1000,
    output_format="parquet"
)

# Step 2: Use the file paths to run SQL queries
file_paths = download_result.get("files", [])
client.query_sql(
    query=f"""
    SELECT 
        to_address as contract_address, 
        COUNT(*) as tx_count,
        SUM(gas_used) as total_gas,
        AVG(gas_used) as avg_gas
    FROM read_parquet('{file_paths[0]}')
    WHERE to_address IS NOT NULL
    GROUP BY to_address
    ORDER BY total_gas DESC
    LIMIT 20
    """,
    files=file_paths
)

组合 SQL 查询流

为了方便起见，您还可以使用处理这两个步骤的组合函数：

# Get top gas-consuming contracts
client.query_blockchain_sql(
    sql_query="""
    SELECT 
        to_address as contract_address, 
        COUNT(*) as tx_count,
        SUM(gas_used) as total_gas,
        AVG(gas_used) as avg_gas
    FROM read_parquet('/path/to/transactions.parquet')
    WHERE to_address IS NOT NULL
    GROUP BY to_address
    ORDER BY total_gas DESC
    LIMIT 20
    """,
    dataset="transactions",
    blocks_from_latest=1000
)

# Find blocks with the most transactions
client.query_blockchain_sql(
    sql_query="""
    SELECT 
        block_number, 
        COUNT(*) as tx_count
    FROM read_parquet('/path/to/transactions.parquet')
    GROUP BY block_number
    ORDER BY tx_count DESC
    LIMIT 10
    """,
    dataset="transactions",
    blocks="15M:16M"
)

# Analyze event logs by topic
client.query_blockchain_sql(
    sql_query="""
    SELECT 
        topic0, 
        COUNT(*) as event_count
    FROM read_parquet('/path/to/logs.parquet')
    GROUP BY topic0
    ORDER BY event_count DESC
    LIMIT 20
    """,
    dataset="logs",
    blocks_from_latest=100
)

注意：对于 SQL 查询，下载数据时请始终使用output_format="parquet"以确保 DuckDB 获得最佳性能。使用query_blockchain_sql时，应使用read_parquet()函数直接在 SQL 中引用文件路径。

使用块范围查询

Cryo MCP 支持 Cryo 的全部块规范语法：

# Using block numbers
client.query_dataset('transactions', blocks='15000000:15001000')

# Using K/M notation
client.query_dataset('logs', blocks='15M:15.01M')

# Using offsets from latest 
client.query_dataset('blocks', blocks_from_latest=100)

合同过滤

按合约地址过滤日志和其他数据：

# Get all logs for USDC contract
client.query_dataset('logs', 
                    blocks='16M:16.1M', 
                    contract='0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48')

列选择

仅包含您需要的列：

# Get just block numbers and timestamps
client.query_dataset('blocks', 
                    blocks='16M:16.1M', 
                    include_columns=['number', 'timestamp'])

发展

项目结构

cryo-mcp/
├── cryo_mcp/           # Main package directory
│   ├── __init__.py     # Package initialization
│   ├── server.py       # Main MCP server implementation
│   ├── sql.py          # SQL query functionality
├── tests/              # Test directory
│   ├── test_*.py       # Test files
├── pyproject.toml      # Project configuration
├── README.md           # Project documentation

运行测试

uv run pytest

执照

麻省理工学院

致谢

基于 Paradigm 出色的Cryo工具构建
使用MCP 协议进行 API 通信

Cryo MCP Server

Integrations