Cryo MCP Server

低温MCP🧊

模型完成协议（MCP）服务器 低温 区块链数据提取工具。

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

面向LLM用户：SQL查询工作流指南

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

1. 下载数据 随着 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

1. 探索架构 随着 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

1. 运行SQL 随着 query_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.

特性

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

安装（可选）

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

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

需求

• Python 3.8+
• 紫外线
• 工作装置 低温
• 访问以太坊RPC端点
• DuckDB（用于SQL查询功能）

快速开始

使用Claude代码

1. 跑 claude mcp add 以获得交互式提示。
2. 进入 uvx 作为运行命令。
3. 进入 cryo-mcp --rpc-url [--data-dir ] 作为args
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 （str，可选）：块范围规范（例如“1000:1010”）
• start_block （int，可选）：开始块编号（块的替代）
• end_block （int，可选）：结束块编号（块的替代）
• use_latest （bool，可选）：如果为True，则查询最新块
• blocks_from_latest （int，可选）：从最新到包含的块数
• contract （str，可选）：要筛选的合约地址
• output_format （str，可选）：输出格式（json、csv、拼花地板）
• 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 （list，可选）：要查询的拼花地板文件路径列表。如果为“无”，将使用数据目录中的所有文件。
• 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()

获取特定拼花地板文件的模式。

参数：

• file_path （str）：拼花地板文件的路径

例子：

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：用于存储下载数据的目录（覆盖CRY_data_DIR环境变量，默认为~/.neo-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() 功能。

使用块范围进行查询

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提供的工具
• 使用 MCP协议 用于API通信