MCP Iceberg

Apache Iceberg MCP 服务器

一个针对Apache Iceberg的全面模型上下文协议（MCP）服务器，使Claude等大型语言模型（LLM）能够高效地与Iceberg数据湖协同工作。该服务器提供了必要的数据探索、质量检查、元数据操作以及性能优化工具。

特点/功能

探索与元数据（6种工具）

• 列出表（或“显示表列表”）列出所有命名空间中的表
• 获取模式（或架构）获取包含类型和元数据的详细架构信息
• 获取分区信息查看分区规范和转换
• 获取表属性访问表配置和属性
• 获取快照列出可用于时间旅行的快照
• 获取表统计信息获取基本统计信息（行数、文件数、大小）

数据质量（6种工具）

• 样本数据获取样本行，可选择随机抽样
• 获取空值计数计算每列中的空值数量及百分比
• 获取唯一计数每列统计不同值的数量
• 获取值分布获取出现次数最多的前N个值及其计数
• 检查重复项基于指定列检测重复行
• 获取列统计信息数值列的统计汇总（最小值、最大值、平均值、标准差）

内容分析（4种工具）

• 预览分区显示现有分区及其大小
• 搜索值查找包含特定值的行
• 获取数据类型概要获取数据类型分布
• 验证模式演变显示模式演进历史

性能优化（3种工具）

• 获取文件统计信息获取数据文件的信息
• 分析偏斜（或“分析偏差”）检测分区不平衡
• 获取表元数据获取完整的元数据以进行优化

实用工具（5种工具）

• 执行查询使用 pandas 查询语法执行查询
• 获取列名获取列名的简单列表
• 检查表是否存在验证表是否存在
• 获取最新快照获取最新的快照详情
• 过滤器预览应用过滤器预览数据

总计：24种工具

要求

• Python 3.12+
• PyIceberg 0.10.0+
• AWS 凭据（用于 Glue 目录）
• 访问Iceberg目录（通过Glue、REST或Hive）

安装

使用紫外线（推荐）

# Install uv if you haven't already
curl -LsSf https://astral.sh/uv/install.sh | sh

# Clone the repository
cd mcpIceberg

# Install dependencies
uv sync

使用 pip

# Create virtual environment
python -m venv .venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate

# Install dependencies
pip install -r requirements.txt

配置

集中式数据库连接

这个项目采用了一个 单一、集中的数据库/目录配置 这一设置在所有工具和操作中是通用的。您只需配置一个数据库连接，所有24个MCP工具将自动使用它。

主要特点：

• ✅ 配置单一入口点 - 无重复连接字符串
• ✅ 所有工具和脚本之间的一致连接
• ✅ 轻松在不同环境（开发/预生产/生产）之间切换
• ✅ 集中的错误处理和连接验证
• ✅ 支持多种目录类型（Glue、REST、Hive）

环境变量

MCP服务器支持多种Iceberg目录类型。使用环境变量进行配置。

快速入门复制模板文件并用您的值进行编辑：

cp .env.sample .env
# Edit .env with your actual credentials

测试您的连接：

python test_connection.py
# or
uv run test_connection.py

看 .env.sample 以获取所有可用配置选项的完整列表，并附有详细解释。

关于集中式配置架构的详细信息，见 DATABASE_CONFIG.md（文件名可译为“数据库配置说明文件”或保持原样，根据上下文决定是否翻译文件名）。

数据库范围限制（安全功能）

你可以将所有操作限制在 单一数据库 为了增强安全性：

# .env
ICEBERG_DATABASE=production

当启用时：

• ✅ 所有操作仅限于指定的数据库
• ✅ 防止意外访问其他数据库
• ✅ 使用简单的表名： users 而不是 production.users
• ✅ 自动阻止跨数据库查询

示例：

# With ICEBERG_DATABASE=production
table = load_table("users")              # ✓ Works - uses production.users
table = load_table("staging.users")      # ✗ Blocked - DatabaseScopeError

关于数据库范围的详细信息，见 \DATABASE_SCOPE.md\ 翻译为中文是：\数据库范围.md\。

AWS Glue 目录（默认）

export ICEBERG_CATALOG_TYPE=glue
export AWS_REGION=us-east-1
export AWS_ACCESS_KEY_ID=your_access_key
export AWS_SECRET_ACCESS_KEY=your_secret_key

REST 目录

export ICEBERG_CATALOG_TYPE=rest
export ICEBERG_REST_URI=http://localhost:8181

Hive 元数据存储（或 Hive 元存储）

export ICEBERG_CATALOG_TYPE=hive
export ICEBERG_HIVE_URI=thrift://localhost:9083

Claude桌面配置

在您的Claude桌面配置文件中添加：

macOS（麦金塔操作系统）: ~/Library/Application Support/Claude/claude_desktop_config.json Windows： %APPDATA%\Claude\claude_desktop_config.json

{
  "mcpServers": {
    "iceberg": {
      "command": "uv",
      "args": [
        "--directory",
        "/path/to/mcpIceberg",
        "run",
        "main.py"
      ],
      "env": {
        "ICEBERG_CATALOG_TYPE": "glue",
        "AWS_REGION": "us-east-1",
        "AWS_ACCESS_KEY_ID": "your_access_key",
        "AWS_SECRET_ACCESS_KEY": "your_secret_key"
      }
    }
  }
}

使用示例

基础探索

List all available tables:
> Use list_tables

Get schema for a specific table:
> Use get_schema with table_name: "my_database.my_table"

Get basic statistics:
> Use get_table_stats with table_name: "my_database.my_table"

数据质量检查

Sample data from a table:
> Use sample_data with table_name: "my_database.my_table", n: 20

Check for null values:
> Use get_null_counts with table_name: "my_database.my_table"

Find duplicates:
> Use check_duplicates with table_name: "my_database.my_table", columns: ["id", "email"]

Get statistical summary:
> Use get_column_stats with table_name: "my_database.my_table"

高级分析

Get value distribution for a column:
> Use get_value_distribution with table_name: "my_database.my_table", column: "status", top_n: 20

Search for specific values:
> Use search_values with table_name: "my_database.my_table", column: "user_id", value: "12345"

Execute custom queries:
> Use execute_query with table_name: "my_database.my_table", query: "age > 30 and status == 'active'"

Preview with filters:
> Use filter_preview with table_name: "my_database.my_table", filters: "amount > 1000", limit: 50

性能分析

Check partition skew:
> Use analyze_skew with table_name: "my_database.my_table"

Get file statistics:
> Use get_file_stats with table_name: "my_database.my_table"

View partition details:
> Use preview_partitions with table_name: "my_database.my_table"

元数据操作

View snapshots for time travel:
> Use get_snapshots with table_name: "my_database.my_table", limit: 20

Check schema evolution:
> Use validate_schema_evolution with table_name: "my_database.my_table"

Get table properties:
> Use get_table_properties with table_name: "my_database.my_table"

表名格式

所有接受（某种输入或条件）的工具 table_name 参数需要a 完全限定名 格式为：

namespace.table_name

示例：

• production.users
• analytics.daily_metrics
• staging.orders

响应格式

所有工具均以JSON格式返回响应。示例响应来自 get_schema：

{
  "table": "my_database.my_table",
  "schema_id": 1,
  "column_count": 5,
  "columns": [
    {
      "name": "id",
      "type": "long",
      "nullable": false,
      "id": 1
    },
    {
      "name": "name",
      "type": "string",
      "nullable": true,
      "id": 2
    }
  ]
}

错误处理

服务器能够优雅地处理错误，并返回包含信息的错误消息：

• 未找到表验证表名是否为完全限定名
• 连接错误检查您的目录配置和凭据
• 权限错误确保您的AWS凭证具有适当的权限
• 查询错误验证您的查询语法（用于过滤的 pandas 查询格式）

查询语法

这个（或：该） execute_query 并且 filter_preview 工具使用 pandas 查询语法：

示例：

# Numeric comparisons
"age > 30"
"amount >= 1000 and amount  18"

# Complex conditions
"(age > 30 or vip == True) and status == 'active'"

性能考量

• 抽样对于大型表格，请使用 sample_data 带有小的 n 避免加载整个表的值
• 列选择当可能时，在诸如……之类的工具中指定列 get_null_counts 减少数据传输
• 限制大多数工具都设定了合理的默认限制，以防止加载过多数据
• 随机抽样随机抽样加载了请求行数的10倍，以提高随机性

记录（日志）

日志被写入标准错误输出（stderr），并包含：

• 带有参数的工具调用
• 目录初始化状态
• 带有堆栈跟踪的错误详情

在 Claude Desktop 开发者控制台中查看日志，或在手动运行时检查调试输出。

发展

手动运行

# With environment variables
export ICEBERG_CATALOG_TYPE=glue
export AWS_REGION=us-east-1

uv run main.py

测试

# Run tests
uv run pytest tests/

# Run with coverage
uv run pytest --cov=main tests/

添加新工具

1. 将异步方法添加到 IcebergMCPServer 班级；课程（根据上下文，"class" 可以翻译为“班级”或“课程”）
2. 将工具定义添加到 get_tools() 方法
3. 处理程序将自动将调用路由到您的方法
4. 返回一个将被JSON序列化的字典

示例：

async def my_new_tool(self, table_name: str, param: str) -> Dict[str, Any]:
    """Tool description"""
    try:
        table = self._load_table(table_name)
        # Your logic here
        return {
            "table": table_name,
            "result": "value"
        }
    except Exception as e:
        logger.error(f"Error in my_new_tool: {e}")
        raise

故障排除

MCP服务器未在Claude桌面显示

1. 检查配置文件路径和JSON语法
2. 完全重启Claude桌面版
3. 检查Claude Desktop的日志以查找错误

AWS Glue 连接问题

1. 验证AWS凭证是否有效
2. 检查IAM权限（需要Glue和S3访问权限）
3. 验证AWS区域是否正确
4. 使用 AWS CLI 测试凭证： aws glue get-databases

表未找到错误

1. 验证表名格式： namespace.table_name
2. 检查表是否存在：使用 list_tables 首先
3. 验证访问命名空间的权限

性能问题

1. 使用较小的样本量
2. 在可能的情况下指定列以减少数据传输
3. 在可用时使用分区过滤器
4. 考虑使用 approximate 用于计算不同计数的参数

建筑

Claude Desktop/CLI
    �
MCP Protocol (stdio)
    �
IcebergMCPServer
    �
PyIceberg Library
    �
Iceberg Catalog (Glue/REST/Hive)
    �
Iceberg Tables (S3/HDFS/etc)

贡献

欢迎投稿！请：

1. 为仓库创建分支（或“克隆”仓库）
2. 创建一个特性分支
3. 为新功能添加测试
4. 确保所有测试通过
5. 提交一个拉取请求

许可证

这个项目是开源的，并且遵循MIT许可证发布。

支持

对于问题和疑问：

• GitHub Issues（GitHub问题） 创建一个问题
• 文档：本README文件
• PyIceberg 文档：https://py.iceberg.apache.org/

更新日志

v0.1.0（初始发布）

• 在5个类别中实施了24种工具
• 支持Glue、REST和Hive目录
• 全面的错误处理
• JSON 响应格式
• 完整的表元数据和模式操作
• 数据质量检查与抽样
• 性能优化工具
• 内容分析与搜索功能

致谢

• 使用……构建 PyIceberg（注：此名称可能是一个特定项目或库的名称，直接翻译为中文可能无实际意义，因此保持原样）
• 使用 模型上下文协议