Cryo MCP 🧊

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

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

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

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

1. 下载数据，使用 query_dataset：

   result = query_dataset(
       dataset="blocks",  # 或 "transactions", "logs" 等
       blocks="15000000:15001000",  # 或使用 blocks_from_latest=100
       output_format="parquet"  # 重要：SQL 使用 parquet 格式
   )
   files = result.get("files", [])  # 获取返回的文件路径
   

2. 探索模式，使用 get_sql_table_schema：

   # 检查文件中可用的列
   schema = get_sql_table_schema(files[0])
   # 现在你可以看到所有的列、数据类型和示例数据
   

3. 运行 SQL，使用 query_sql：

   # 选项 1：简单表引用（DuckDB 将表名与文件匹配）
   sql_result = query_sql(
       query="SELECT block_number, timestamp, gas_used FROM blocks",
       files=files  # 传递步骤 1 中的文件
   )
   
   # 选项 2：使用 read_parquet() 和显式文件路径
   sql_result = query_sql(
       query=f"SELECT block_number, timestamp, gas_used FROM read_parquet('{files[0]}')",
       files=files  # 传递步骤 1 中的文件
   )
   

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

# 选项 1：简单表引用
result = query_blockchain_sql(
    sql_query="SELECT * FROM blocks",
    dataset="blocks",
    blocks_from_latest=100
)

# 选项 2：使用 read_parquet()
result = query_blockchain_sql(
    sql_query="SELECT * FROM read_parquet('/path/to/file.parquet')",  # 路径无关紧要
    dataset="blocks",
    blocks_from_latest=100
)

有关完整的可运行示例，请参见 examples/sql_workflow_example.py。

特性

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

安装（可选）

如果您将直接使用 uvx 运行该工具，则不需要此操作。

# 使用 UV 安装（推荐）
uv tool install cryo-mcp

要求

• Python 3.8+
• uv
• Cryo 的有效安装
• 访问以太坊 RPC 端点
• DuckDB（用于 SQL 查询功能）

快速开始

与 Claude Code 一起使用

1. 运行 claude mcp add 以获得交互式提示。
2. 输入 uvx 作为要运行的命令。
3. 输入 cryo-mcp --rpc-url <ETH_RPC_URL> [--data-dir <DATA_DIR>] 作为参数。
4. 或者，提供 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, optional)：区块范围规范（例如，“1000:1010”）
• start_block (int, optional)：起始区块号（替代 blocks）
• end_block (int, optional)：结束区块号（替代 blocks）
• use_latest (bool, optional)：如果为 True，则查询最新区块
• blocks_from_latest (int, optional)：要包含的最新区块之前的区块数
• contract (str, optional)：要按其过滤的合约地址
• output_format (str, optional)：输出格式（“json”、“csv”、“parquet”）
• include_columns (list, optional)：要包含在默认值旁边的列
• exclude_columns (list, optional)：要从默认值中排除的列

示例：

# 获取区块 15M 到 15.01M 的交易
client.query_dataset('transactions', blocks='15M:15.01M')

# 获取特定合约的最新 100 个区块的日志
client.query_dataset('logs', blocks_from_latest=100, contract='0x1234...')

# 仅获取最新区块
client.query_dataset('blocks', use_latest=True)

lookup_dataset()

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

参数：

• name (str)：要查找的数据集的名称
• sample_start_block (int, optional)：示例数据的起始区块
• sample_end_block (int, optional)：示例数据的结束区块
• use_latest_sample (bool, optional)：使用最新区块作为示例
• sample_blocks_from_latest (int, optional)：示例的最新区块之前的区块数

示例：

client.lookup_dataset('logs')

get_latest_ethereum_block()

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

示例：

client.get_latest_ethereum_block()

SQL 查询工具

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

query_sql()

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

参数：

• query (str)：要执行的 SQL 查询
• files (list, optional)：要查询的 parquet 文件路径列表。 如果为 None，将使用数据目录中的所有文件。
• include_schema (bool, optional)：是否在结果中包含模式信息

示例：

# 对所有可用文件运行
client.query_sql("SELECT * FROM read_parquet('/path/to/blocks.parquet') LIMIT 10")

# 对特定文件运行
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, optional)：要查询的数据集（例如，“blocks”、“transactions”）
• blocks (str, optional)：区块范围规范
• start_block (int, optional)：起始区块号
• end_block (int, optional)：结束区块号
• use_latest (bool, optional)：如果为 True，则查询最新区块
• blocks_from_latest (int, optional)：要包含的最新区块之前的区块数
• contract (str, optional)：要按其过滤的合约地址
• force_refresh (bool, optional)：即使数据存在也强制下载新数据
• include_schema (bool, optional)：在结果中包含模式信息

示例：

# 如果需要，自动下载区块数据，然后运行 SQL 查询
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 查询流程

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

# 步骤 1：下载数据并获取文件路径
download_result = client.query_dataset(
    dataset="transactions",
    blocks_from_latest=1000,
    output_format="parquet"
)

# 步骤 2：使用文件路径运行 SQL 查询
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 查询流程

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

# 获取 gas 消耗最多的合约
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
)

# 查找交易最多的区块
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"
)

# 按主题分析事件日志
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 的完整区块规范语法：

# 使用区块号
client.query_dataset('transactions', blocks='15000000:15001000')

# 使用 K/M 表示法
client.query_dataset('logs', blocks='15M:15.01M')

# 使用最新区块的偏移量
client.query_dataset('blocks', blocks_from_latest=100)

合约过滤

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

# 获取 USDC 合约的所有日志
client.query_dataset('logs', 
                    blocks='16M:16.1M', 
                    contract='0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48')

列选择

仅包含您需要的列：

# 仅获取区块号和时间戳
client.query_dataset('blocks', 
                    blocks='16M:16.1M', 
                    include_columns=['number', 'timestamp'])

开发

项目结构

cryo-mcp/
├── cryo_mcp/           # 主包目录
│   ├── __init__.py     # 包初始化
│   ├── server.py       # 主 MCP 服务器实现
│   ├── sql.py          # SQL 查询功能
├── tests/              # 测试目录
│   ├── test_*.py       # 测试文件
├── pyproject.toml      # 项目配置
├── README.md           # 项目文档

运行测试

uv run pytest

许可证

MIT

鸣谢

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