内容
# Cryo MCP 🧊
Cryo 区块链数据提取工具的模型完成协议（MCP）服务器。
Cryo MCP 允许您通过实现 MCP 协议的 API 服务器访问 Cryo 强大的区块链数据提取功能，从而可以轻松地从任何兼容 MCP 的客户端查询区块链数据。
## For LLM Users: SQL Query Workflow Guide
使用此 MCP 服务器在区块链数据上运行 SQL 查询时，请遵循此工作流程：
1. **下载数据** 使用 `query_dataset`：
```python
result = query_dataset(
dataset="blocks", # 或 "transactions", "logs" 等
blocks="15000000:15001000", # 或使用 blocks_from_latest=100
output_format="parquet" # 重要：使用 parquet 进行 SQL 查询
)
files = result.get("files", []) # 获取返回的文件路径
```
2. **探索模式** 使用 `get_sql_table_schema`：
```python
# 检查文件中可用的列
schema = get_sql_table_schema(files[0])
# 现在您可以查看所有列、数据类型和示例数据
```
3. **运行 SQL** 使用 `query_sql`：
```python
# 选项 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` 结合两种方法：
```python
# 选项 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](examples/sql_workflow_example.py)。
## 功能
- **完整的 Cryo 数据集访问**：通过 API 服务器查询任何 Cryo 数据集
- **MCP 集成**：与 MCP 客户端无缝集成
- **灵活的查询选项**：支持所有主要的 Cryo 过滤和输出选项
- **区块范围选项**：查询特定区块、最新区块或相对范围
- **合约过滤**：按合约地址过滤数据
- **最新区块访问**：轻松访问最新以太坊区块数据
- **多种输出格式**：支持 JSON、CSV 和 Parquet
- **模式信息**：获取详细的 dataset 模式和示例数据
- **SQL 查询**：直接对下载的区块链数据运行 SQL 查询
## 安装（可选）
如果您将使用 `uvx` 直接运行该工具，则不需要此步骤。
```bash
# 使用 UV 安装（推荐）
uv tool install cryo-mcp
```
## 要求
- Python 3.8+
- uv
- [Cryo](https://github.com/paradigmxyz/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 数据集的列表。
示例：
```python
client.list_datasets()
```
### `query_dataset()`
使用各种过滤选项查询 Cryo 数据集。
参数：
- `dataset`（str）：要查询的数据集名称（例如，'blocks'，'transactions'，'logs'）
- `blocks`（str，可选）：区块范围规范（例如，'1000:1010'）
- `start_block`（int，可选）：起始区块号（替代 `blocks`）
- `end_block`（int，可选）：结束区块号（替代 `blocks`）
- `use_latest`（bool，可选）：如果为 True，则查询最新区块
- `blocks_from_latest`（int，可选）：从最新区块开始的区块数
- `contract`（str，可选）：合约地址，用于过滤
- `output_format`（str，可选）：输出格式（'json'，'csv'，'parquet'）
- `include_columns`（list，可选）：除默认值外要包含的列
- `exclude_columns`（list，可选）：从默认值中排除的列
示例：
```python
# 获取区块 15M 至 15.01M 的交易
client.query_dataset('transactions', blocks='15M:15.01M')
# 获取最新 100 个区块的日志，适用于特定合约
client.query_dataset('logs', blocks_from_latest=100, contract='0x1234...')
```
### `lookup_dataset()`
获取有关特定数据集的详细信息，包括模式和示例数据。
参数：
- `name`（str）：要查找的数据集名称
- `sample_start_block`（int，可选）：样本数据的起始区块
- `sample_end_block`（int，可选）：样本数据的结束区块
- `use_latest_sample`（bool，可选）：使用最新区块的样本
- `sample_blocks_from_latest`（int，可选）：从最新区块开始的样本区块数
示例：
```python
client.lookup_dataset('logs')
```
### `get_latest_ethereum_block()`
返回最新以太坊区块的信息。
示例：
```python
client.get_latest_ethereum_block()
```
### SQL 查询工具
Cryo MCP 包括多个用于对区块链数据运行 SQL 查询的工具：
### `query_sql()`
对下载的区块链数据运行 SQL 查询。
参数：
- `query`（str）：要执行的 SQL 查询
- `files`（list，可选）：要查询的 parquet 文件路径列表。如果为 None，则使用数据目录中的所有文件。
- `include_schema`（bool，可选）：是否在结果中包含模式信息
示例：
```python
# 对所有可用文件运行查询
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，可选）：要查询的数据集（例如，'blocks'，'transactions'）
- `blocks`（str，可选）：区块范围规范
- `start_block`（int，可选）：起始区块号
- `end_block`（int，可选）：结束区块号
- `use_latest`（bool，可选）：如果为 True，则查询最新区块
- `blocks_from_latest`（int，可选）：从最新区块开始的区块数
- `contract`（str，可选）：合约地址，用于过滤
- `force_refresh`（bool，可选）：强制下载新数据，即使存在
- `include_schema`（bool，可选）：在结果中包含模式信息
示例：
```python
# 自动下载 blocks 数据（如果需要），然后运行 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 查询的表。
示例：
```python
client.list_available_sql_tables()
```
### `get_sql_table_schema()`
获取特定 parquet 文件的模式。
参数：
- `file_path`（str）：parquet 文件的路径
示例：
```python
client.get_sql_table_schema("/path/to/blocks.parquet")
```
### `get_sql_examples()`
获取不同区块链数据集的 SQL 查询示例。
示例：
```python
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 查询流程
您可以将数据提取和查询分成两个单独的步骤：
```python
# 步骤 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 查询流程
为了方便，您也可以使用组合函数来处理这两个步骤：
```python
# 获取消耗 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` 时，您应该在 SQL 中直接引用文件路径，使用 `read_parquet()` 函数。
### 查询区块范围
Cryo MCP 支持 Cryo 的完整区块规范语法：
```python
# 使用区块编号
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)
```
### 合约过滤
按合约地址过滤日志和其他数据：
```python
# 获取 USDC 合约的所有日志
client.query_dataset('logs',
blocks='16M:16.1M',
contract='0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48')
```
### 列选择
仅包含需要的列：
```python
# 获取区块编号和时间戳
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
## 致谢
- 在 [Cryo](https://github.com/paradigmxyz/cryo) 工具的基础上构建，由 Paradigm 团队开发
- 使用 [MCP 协议](https://github.com/mcp-team/mcp) 进行 API 通信
