MCP Iceberg Catalog

一个用于与 Apache Iceberg 交互的 MCP (模型上下文协议) 服务器实现。该服务器通过 Claude desktop 提供一个 SQL 接口，用于查询和管理 Iceberg 表。

Claude Desktop 作为你的 Iceberg 数据湖 Catalog

如何在 Claude Desktop 中安装

通过 Smithery 安装

要通过 Smithery 为 Claude Desktop 自动安装 MCP Iceberg Catalog：

npx -y @smithery/cli install @ahodroj/mcp-iceberg-service --client claude

1. 前提条件

   • Python 3.10 或更高版本
   • UV 包管理器（推荐）或 pip
   • 访问 Iceberg REST catalog 和 S3 兼容的存储

2. 如何在 Claude Desktop 中安装 将以下配置添加到 claude_desktop_config.json：

{
  "mcpServers": {
    "iceberg": {
      "command": "uv",
      "args": [
        "--directory",
        "PATH_TO_/mcp-iceberg-service",
        "run",
        "mcp-server-iceberg"
      ],
      "env": {
        "ICEBERG_CATALOG_URI" : "http://localhost:8181",
        "ICEBERG_WAREHOUSE" : "你的 ICEBERG 仓库名称",
        "S3_ENDPOINT" : "如果使用 S3，则为可选",
        "AWS_ACCESS_KEY_ID" : "你的 S3 访问密钥",
        "AWS_SECRET_ACCESS_KEY" : "你的 S3 密钥"
      }
    }
  }
}

设计

架构

MCP 服务器构建在三个主要组件之上：

1. MCP 协议处理器

   • 实现用于与 Claude 通信的模型上下文协议
   • 通过 stdio 处理请求/响应周期
   • 管理服务器生命周期和初始化

2. 查询处理器

   • 使用 sqlparse 解析 SQL 查询
   • 支持的操作：
     • LIST TABLES
     • DESCRIBE TABLE
     • SELECT
     • INSERT

3. Iceberg 集成

   • 使用 pyiceberg 进行表操作
   • 与 PyArrow 集成以实现高效的数据处理
   • 管理 catalog 连接和表操作

PyIceberg 集成

服务器以多种方式利用 PyIceberg：

1. Catalog 管理

   • 连接到 REST catalogs
   • 管理表元数据
   • 处理命名空间操作

2. 数据操作

   • 在 PyIceberg 和 PyArrow 类型之间转换
   • 通过 PyArrow 表处理数据插入
   • 管理表模式和字段类型

3. 查询执行

   • 将 SQL 转换为 PyIceberg 操作
   • 处理数据扫描和过滤
   • 管理结果集转换

需要进一步实现

1. 查询操作

   • [ ] 实现 UPDATE 操作
   • [ ] 添加 DELETE 支持
   • [ ] 支持使用模式定义创建表 (CREATE TABLE)
   • [ ] 添加 ALTER TABLE 操作
   • [ ] 实现表分区支持

2. 数据类型

   • [ ] 支持复杂类型（数组、映射、结构体）
   • [ ] 添加带时区的时间戳处理
   • [ ] 支持 decimal 类型
   • [ ] 添加嵌套字段支持

3. 性能改进

   • [ ] 实现批量插入
   • [ ] 添加查询优化
   • [ ] 支持并行扫描
   • [ ] 添加缓存层以用于频繁访问的数据

4. 安全特性

   • [ ] 添加身份验证机制
   • [ ] 实现基于角色的访问控制
   • [ ] 添加行级安全性
   • [ ] 支持加密连接

5. 监控和管理

   • [ ] 添加指标收集
   • [ ] 实现查询日志记录
   • [ ] 添加性能监控
   • [ ] 支持表维护操作

6. 错误处理

   • [ ] 改进错误消息
   • [ ] 为瞬时故障添加重试机制
   • [ ] 实现事务支持
   • [ ] 添加数据验证