ssb-mcp-server

🚀 SSB MCP 服务器

SSB MCP 服务器是一款模型上下文协议服务器，它能全面接入 SQL Stream Builder (SSB)，既支持直接访问 SSB，也支持与 Apache Knox 集成。无论是独立部署的 SSB，还是 Cloudera Data Platform (CDP) 中的 SSB 部署，该服务器都能通过 Claude Desktop 提供对 SSB 功能的完整访问。

主 SSB 界面展示了包含可用流和作业的主页仪表盘，突出展示了 SSB MCP 服务器的集成及品牌标识。

🚀 快速开始

独立 SSB 部署（使用 Docker Compose）

1. 启动 SSB 服务：

   git clone https://github.com/your-org/ssb-mcp-server.git
   cd ssb-mcp-server
   docker-compose up -d
   

2. 配置 Claude Desktop：编辑 ~/Library/Application Support/Claude/claude_desktop_config.json 文件：

   {
     "mcpServers": {
       "ssb-mcp-server": {
         "command": "/FULL/PATH/TO/SSB-MCP-Server/run_mcp_server.sh",
         "args": [],
         "cwd": "/FULL/PATH/TO/SSB-MCP-Server"
       }
     }
   }
   

3. 重启 Claude Desktop，即可开始与 SSB 流进行交互！

CDP SSB 部署

SSB API 的基础 URL 通常为：

https://<your-ssb-host>/ssb/api/v1

从 CDP 界面获取 Knox JWT 令牌，并在以下配置中使用。

✨ 主要特性

• 多种认证方式：
  • 直接 SSB 认证：适用于独立 SSB 部署的基本认证方式。
  • Knox 集成：适用于 CDP 部署，支持使用承载令牌、Cookie 和密码令牌进行认证。
• 默认只读模式：可安全地探索 SSB 流和配置。
• 全面覆盖 SSB API：拥有 80 多种 MCP 工具，可对 SSB 进行全面管理，涵盖以下方面：
  • 高级作业管理：包括事件历史记录、状态管理、作业复制、数据源克隆等功能。
  • 监控与诊断：可查看系统健康状况、性能计数器和进行 SQL 分析。
  • 增强型表管理：提供详细的表信息、层次结构、验证和创建功能。
  • 连接器与格式管理：管理数据格式、连接器详细信息和 JAR 文件信息。
  • 用户与项目管理：涉及用户设置、项目管理、用户信息和项目创建等操作。
  • API 密钥管理：包括密钥生命周期管理、创建、删除和查看详细信息。
  • 环境管理：支持环境切换、配置和创建。
  • 同步与配置：实现项目导出/导入、同步管理和验证。
  • UDF 管理：涵盖 UDF 生命周期管理、执行、工件和自定义函数。
  • 流管理：可列出、创建、更新、删除、启动和停止 SQL 流。
  • 查询执行：执行 SQL 查询并通过采样获取实时结果。
  • 样本数据访问：从运行的作业中检索流数据样本。
  • 作业管理：监控作业状态、获取作业详细信息和管理作业生命周期。
  • 模式发现：探索表模式和可用表。
  • 函数管理：列出和检查用户定义的函数。
  • 连接器管理：探索可用的连接器。
  • Kafka 集成：列出和检查 Kafka 主题。
  • 集群监控：获取集群信息和健康状态。
  • 性能指标：监控流性能和指标。

📦 安装指南

选项 1：Claude Desktop（本地安装）

1. 克隆并安装：

   git clone https://github.com/your-org/ssb-mcp-server.git
   cd ssb-mcp-server
   python3 -m venv .venv
   source .venv/bin/activate
   pip install -e .
   

2. 配置 Claude Desktop：编辑 ~/Library/Application Support/Claude/claude_desktop_config.json 文件：

    {
      "mcpServers": {
        "ssb-mcp-server": {
          "command": "/FULL/PATH/TO/SSB-MCP-Server/.venv/bin/python",
          "args": [
            "-m",
            "ssb_mcp_server.server"
          ],
          "env": {
            "MCP_TRANSPORT": "stdio",
            "SSB_API_BASE": "https://ssb-gateway.yourshere.cloudera.site/ssb/api/v1",
            "KNOX_TOKEN": "<your_knox_bearer_token>",
            "SSB_READONLY": "true"
          }
        }
      }
    }
   

3. 重启 Claude Desktop，即可开始询问关于 SSB 流的问题！

选项 2：直接安装（Cloudera Agent Studio）

若要在 Cloudera Agent Studio 中使用，可使用 uvx 命令：

{
  "mcpServers": {
    "ssb-mcp-server": {
      "command": "uvx",
      "args": [
        "--from",
        "git+https://github.com/your-org/ssb-mcp-server@main",
        "run-server"
      ],
      "env": {
        "MCP_TRANSPORT": "stdio",
        "SSB_API_BASE": "https://ssb-gateway.yourshere.cloudera.site/ssb/api/v1",
        "KNOX_TOKEN": "<your_knox_bearer_token>",
        "SSB_READONLY": "true"
      }
    }
  }
}

🔧 配置选项

所有配置均通过环境变量完成：

直接 SSB 认证（独立部署）

| 变量 | 是否必需 | 描述 | |------|----------|-------------| | SSB_API_BASE | 是 | 完整的 SSB API URL（例如：http://localhost:18121） | | SSB_USER | 是 | SSB 用户名（例如：admin） | | SSB_PASSWORD | 是 | SSB 密码（例如：admin） | | SSB_READONLY | 否 | 只读模式（默认：false） | | TIMEOUT_SECONDS | 否 | HTTP 超时时间（秒）（默认：30） |

Knox 认证（CDP）

| 变量 | 是否必需 | 描述 | |------|----------|-------------| | KNOX_GATEWAY_URL | 是* | Knox 网关 URL（例如：https://host:8444/gateway/ssb） | | KNOX_TOKEN | 是* | 用于认证的 Knox JWT 令牌 | | KNOX_COOKIE | 否 | 可选：提供完整的 Cookie 字符串代替令牌 | | KNOX_PASSCODE_TOKEN | 否 | 可选：Knox 密码令牌（自动交换为 JWT） | | KNOX_USER | 否 | 用于基本认证的 Knox 用户名 | | KNOX_PASSWORD | 否 | 用于基本认证的 Knox 密码 | | KNOX_VERIFY_SSL | 否 | 验证 SSL 证书（默认：true） | | KNOX_CA_BUNDLE | 否 | CA 证书包的路径 | | SSB_READONLY | 否 | 只读模式（默认：true） | | TIMEOUT_SECONDS | 否 | HTTP 超时时间（秒）（默认：30） |

* SSB_API_BASE（直接认证）或 KNOX_GATEWAY_URL（Knox 认证）二者必须提供其一。

💻 使用示例

基础用法

配置完成后，你可以向 Claude 提出以下问题：

基本信息

• "我正在运行的 SSB 版本是什么？"
• "列出我所有的 SQL 流"
• "有哪些表可供查询？"
• "有哪些可用的连接器？"
• "列出所有 Kafka 主题"
• "集群的健康状况如何？"

Claude 可以显示 SSB 环境中所有可用的表，包括内置表和自定义表。

查询执行与数据访问

• "执行此查询：SELECT * FROM NVDA"
• "创建一个采样所有消息的作业：SELECT * FROM NVDA"
• "显示作业 1234 的状态"

Claude 可以列出所有正在运行的 SSB 作业，显示其状态、创建时间和详细信息。

• "从作业 1234 获取样本数据"
• "列出所有作业及其样本信息"
• "显示 ReadNVDA 作业的最新数据"

作业管理与控制

• "停止作业 1234"
• "以采样所有消息的方式重启作业 1234"
• "以自定义采样（500ms 间隔）重启作业 1234"
• "将作业 1234 的采样配置为采样所有消息"

Claude 可以通过执行 SQL 查询创建新的 SSB 作业，具备完整的作业管理功能。

流管理

• "创建一个名为 'sales_analysis' 的新流，使用此 SQL：SELECT * FROM sales WHERE amount > 1000"
• "显示 'user_events' 流的详细信息"
• "我的 'sales_stream' 状态如何？"

Kafka 表管理

• "从主题 'user-events' 创建一个名为 'user_events' 的本地 Kafka 表"
• "在 Flink 目录中注册一个 Kafka 表，使其可查询"
• "创建一个使用 JSON 格式的本地 Kafka 表"

Claude 可以创建连接到 Kafka 主题的新虚拟表，并进行适当的模式和连接器配置。

• "验证 'local-kafka' 是否为有效的连接器类型"
• "创建一个用于实时数据流的虚拟表"

高级作业管理

• "显示作业 1234 的事件历史记录"
• "获取作业 1234 的详细状态"
• "复制作业 1234 以创建一个新作业"
• "克隆表 'user_events' 的数据源"
• "获取作业 1234 的物化视图端点"
• "为作业 1234 创建一个物化视图端点"

物化视图

• "获取作业 1234 的物化视图端点"
• "为作业 1234 创建一个物化视图端点"

⚠️ 重要限制：物化视图（MVs）必须通过 SSB 界面创建。MCP 服务器可以从现有的物化视图中检索数据，但不能以编程方式创建新的物化视图。若要创建物化视图，请按以下步骤操作：

1. 在 SSB 界面中导航到你的作业。
2. 转到物化视图部分。
3. 通过界面配置并创建 MV。
4. 使用 MCP 服务器查询已创建的 MV 数据。

监控与诊断

• "检查系统的心跳和健康状况"
• "显示诊断计数器"
• "分析此 SQL 查询的性能：SELECT * FROM NVDA WHERE close > 100"
• "当前系统性能如何？"

增强型表管理

• "显示所有表的详细信息"
• "按目录获取表的层次结构"
• "验证此数据源配置"
• "使用完整配置创建一个新表"
• "获取表 'user_events' 的详细信息"

Claude 可以提供特定表的详细信息，包括其模式和配置。

用户与项目管理

• "显示我的用户设置和偏好"
• "将我的用户设置更新为启用暗模式"
• "列出所有可用的项目"
• "创建一个名为 'analytics' 的新项目"
• "获取项目 'ffffffff' 的详细信息"
• "显示我的用户信息"

API 密钥管理

• "列出我所有的 API 密钥"
• "创建一个用于外部访问的新 API 密钥"
• "删除 API 密钥 'key123'"
• "获取 API 密钥 'key123' 的详细信息"

环境管理

• "列出所有可用的环境"
• "切换到环境 'production'"
• "创建一个名为 'staging' 的新环境"
• "获取环境 'dev' 的详细信息"
• "停用当前环境"

同步与配置

• "显示当前的同步配置"
• "更新用于 Git 集成的同步配置"
• "导出项目 'analytics' 的配置"
• "从 Git 导入项目配置"
• "验证项目 'test' 的同步配置"

UDF 管理

• "列出所有用户定义的函数并显示详细信息"
• "使用参数运行 UDF 'custom_aggregate'"
• "创建一个用于数据转换的新 UDF"
• "更新 UDF 'my_function' 的配置"
• "获取 UDF 工件和依赖项"

样本数据示例

MCP 服务器可以通过不同的采样模式检索实时流数据样本：

Claude 可以从运行的作业中检索实时样本数据，显示实际的流数据。

定期采样（默认）：

{
  "records": [
    {
      "___open": "185.0919",
      "___high": "185.1200", 
      "___low": "184.9400",
      "___close": "184.9700",
      "___volume": "61884",
      "eventTimestamp": "2025-10-08T18:34:10.915Z"
    }
  ],
  "job_status": "RUNNING",
  "end_of_samples": false,
  "message": "Retrieved 1 sample records"
}

采样所有消息模式：

{
  "sampling_mode": "sample_all_messages",
  "sample_interval": 0,
  "sample_count": 10000,
  "window_size": 10000,
  "message": "Job created with comprehensive sampling enabled"
}

高级用法

采样所有消息

若要进行全面的数据采样，可使用 sample_all_messages=True 选项：

# 创建作业并采样所有消息
execute_query_with_sampling("SELECT * FROM NVDA", sample_all_messages=True)

# 以采样所有消息的方式重启作业
restart_job_with_sampling(1234, "SELECT * FROM NVDA", sample_all_messages=True)

配置：

• sample_interval: 0（立即采样）
• sample_count: 10000（高计数以捕获所有消息）
• window_size: 10000（大窗口以进行全面采样）

自定义采样配置

可根据具体需求微调采样行为：

# 以 500ms 间隔进行自定义采样
execute_query_with_sampling("SELECT * FROM NVDA", 
                           sample_interval=500, 
                           sample_count=500, 
                           window_size=500)

# 配置现有作业的采样
configure_sampling("sample_id", 
                  sample_interval=200, 
                  sample_count=1000, 
                  window_size=1000)

作业管理

可进行完整的作业生命周期管理：

# 带保存点停止作业
stop_job(1234, savepoint=True)

# 用新的 SQL 重启作业
execute_job(1234, "SELECT * FROM NEW_TABLE")

# 带采样选项重启作业
restart_job_with_sampling(1234, "SELECT * FROM NVDA", 
                         sample_interval=1000, 
                         sample_all_messages=False)

Kafka 表创建

创建仅限制使用本地 Kafka 连接器的表：

# 步骤 1：创建数据源（创建配置）
create_kafka_table("user_events", "user-events")  # 默认使用 local-kafka

# 步骤 2：在 Flink 目录中注册表（使其可查询）
register_kafka_table("user_events", "user-events")  # 在 ssb.ssb_default 中创建 ssb_user_events（回退到 default_catalog.ssb_default）

# 高级用法：自定义模式注册
custom_schema = [
    {"name": "id", "type": "STRING"},
    {"name": "name", "type": "STRING"},
    {"name": "timestamp", "type": "TIMESTAMP"}
]
register_kafka_table("custom_table", "custom-topic", custom_schema)  # 创建 ssb_custom_table

# 不使用 ssb_ 前缀
register_kafka_table("raw_data", "raw-topic", use_ssb_prefix=False)  # 创建 raw_data（无前缀）

# 自定义目录和数据库
register_kafka_table("custom_table", "custom-topic", catalog="default_catalog", database="default_database")

# 带自定义设置的本地 Kafka
create_kafka_table("local_data", "local-topic", "local-kafka", "localhost:9092", "json", "earliest-offset")
register_kafka_table("local_data", "local-topic")  # 创建 ssb_local_data

# 验证连接器类型
validate_kafka_connector("local-kafka")  # 返回验证详细信息
validate_kafka_connector("kafka")  # 返回错误 - 仅允许 local-kafka

两步流程：创建 Kafka 表需要两个步骤：

1. 创建数据源：使用 create_kafka_table() 创建数据源配置。
2. 在目录中注册：使用 register_kafka_table() 使表可查询。

自动注册：register_kafka_table() 函数具有以下特点：

• 使用 DDL 在 Flink 目录中注册表（默认：ssb.ssb_default，若 ssb 目录不可用则回退到 default_catalog.ssb_default）。
• 可配置目录和数据库参数，实现灵活的命名空间控制。
• 当请求的目录不可用时自动回退。
• 默认在与现有表（如 NVDA）相同的数据库中创建表。
• 自动为表名添加 ssb_ 前缀（可配置）。
• 根据主题数据自动创建模式。
• 通过检查正确的数据库上下文验证表是否可查询。
• 返回包含完整表名、目录和数据库信息的成功注册确认。

命名约定：

• 默认：表名自动添加 ssb_ 前缀（例如：user_events → ssb_user_events）。
• 覆盖：使用 use_ssb_prefix=False 禁用前缀。
• 现有表：已以 ssb_ 开头的表名不会被修改。

命名空间配置：

• 默认：表在 ssb.ssb_default 命名空间中创建（若 ssb 目录不可用则回退到 default_catalog.ssb_default）。
• 自定义目录：使用 catalog 参数指定不同的目录。
• 自定义数据库：使用 database 参数指定不同的数据库。
• 自动回退：若请求的目录不可用，系统会自动回退到 default_catalog。
• 完全控制：目录和数据库均可配置，提供最大的灵活性。

验证：

• 使用 SHOW TABLES; 确认表是否可查询。
• 表在 default_catalog.ssb_default 命名空间中创建（与 NVDA 相同）。
• 可使用完整命名空间（default_catalog.ssb_default.TABLE_NAME）或使用 USE default_catalog.ssb_default; 切换数据库上下文。
• 所有虚拟 Kafka 表与现有表位于同一位置，便于查询。

支持的连接器：

• local-kafka - 本地 Kafka 连接器（虚拟表的唯一选项）。

支持的格式：

• json - JSON 格式（默认）。
• csv - CSV 格式。
• avro - Apache Avro 格式。
• 自定义格式字符串。

📚 详细文档

Docker Compose 设置

仓库中包含一个完整的 Docker Compose 设置，用于本地开发和测试：

包含的服务

• PostgreSQL：用于存储 SSB 元数据的数据库。
• Kafka：消息流平台。
• Flink：流处理引擎。
• NiFi：数据流管理工具。
• Qdrant：向量数据库。
• SSB SSE：SQL Stream Builder 流 SQL 引擎。
• SSB MVE：SQL Stream Builder 物化视图引擎。
• Apache Knox：用于安全访问的网关（可选）。

启动环境

# 启动所有服务
docker-compose up -d

# 检查服务状态
docker-compose ps

# 查看日志
docker-compose logs -f ssb-sse

访问点

• SSB SSE：http://localhost:18121
• SSB MVE：http://localhost:18131
• Flink 作业管理器：http://localhost:8081
• NiFi：http://localhost:8080
• Knox 网关：https://localhost:8444（若启用）

写操作

默认情况下，服务器在 CDP 部署中以只读模式运行，在独立部署中启用写操作。若要更改此设置，请按以下步骤操作：

1. 设置 SSB_READONLY=false（启用写操作）或 SSB_READONLY=true（只读模式）。
2. 重启 MCP 服务器。

写操作包括：

• 创建、更新和删除流。
• 执行创建作业的 SQL 查询。
• 管理作业生命周期（启动、停止、重启）。
• 配置采样参数。
• 作业控制和管理。
• 创建仅支持 Kafka 的表（强制验证）。

全面功能

SSB MCP 服务器现在提供 80 多种 MCP 工具，覆盖 80% 以上的 SSB API，是通过 Claude Desktop 可用的最全面的 SSB 管理平台。

📊 覆盖统计

• MCP 工具总数：80 多个（从 33 个增加）。
• API 覆盖率：80% 以上（从 20% 增加）。
• 功能类别：15 个（从 6 个增加）。
• 可用端点：67 个以上（从 15 个增加）。

🎯 关键功能

• 完整的 SSB 管理：
  • 作业生命周期：创建、监控、控制、复制和管理作业。
  • 数据管理：表、模式、验证和层次结构组织。
  • 系统监控：健康检查、诊断和性能跟踪。
  • 用户管理：设置、项目、环境和 API 密钥。
  • DevOps 集成：同步、导出/导入和配置管理。
• 高级功能：
  • 实时采样：灵活的数据采样，支持“采样所有消息”选项。
  • SQL 分析：在不执行的情况下分析查询以优化性能。
  • 物化视图：创建和管理物化视图端点。
  • 自定义 UDF：用户定义函数管理和执行。
  • 环境控制：支持多环境切换。
  • 项目管理：完整的项目生命周期，支持导出/导入。
• 企业就绪：
  • 安全性：API 密钥管理和用户认证。
  • 监控：全面的系统健康和性能跟踪。
  • 可扩展性：支持多个项目和环境。
  • 集成：Git 同步、配置管理和 DevOps 工作流。
  • 灵活性：可配置的目录、数据库和命名约定。

🚀 使用场景

• 数据工程师：流处理作业管理和监控、实时数据采样和分析、表模式管理和验证、性能优化和故障排除。
• DevOps 工程师：环境管理和配置、项目导出/导入和版本控制、系统监控和健康检查、API 密钥管理和安全。
• 数据科学家：自定义 UDF 开发和执行、数据格式管理和验证、查询分析和优化、实时数据探索。
• 平台管理员：用户和项目管理、系统诊断和监控、连接器和格式管理、同步配置和验证。

测试

SSB MCP 服务器包含一个全面的测试套件。有关详细的测试文档，请参阅 Testing/README.md，其中包括：

• 快速功能测试。
• 覆盖所有 80 多种 MCP 工具的全面测试套件。
• 云 SSB 测试协议。
• 测试配置和最佳实践。
• 详细的测试结果和分析。

快速开始：

cd Testing && python run_tests.py --quick

故障排除

常见问题

1. “Unauthorized” 错误：检查认证凭据。
   • 对于直接 SSB 认证：验证 SSB_USER 和 SSB_PASSWORD。
   • 对于 Knox 认证：验证 KNOX_TOKEN 或 KNOX_USER/KNOX_PASSWORD。
2. “Connection refused” 错误：确保 SSB 服务正在运行。
   • 使用 docker-compose ps 检查服务状态。
   • 验证 docker-compose.yml 中的端口映射。
3. “No sample data available” 错误：作业可能需要时间来生成数据。
   • 使用 get_job_status(job_id) 检查作业状态。
   • 验证作业正在运行并具有采样配置。
   • 尝试使用 sample_all_messages=True 进行全面采样。
4. 作业重启失败：如果作业重启失败：
   • 使用 restart_job_with_sampling() 代替 execute_job()。
   • 检查作业是否处于允许重启的状态。
   • 如果无法重启，请创建一个新作业。
5. SSL 证书错误：对于 Knox 部署：
   • 对于自签名证书，设置 KNOX_VERIFY_SSL=false。
   • 或者使用 KNOX_CA_BUNDLE 提供适当的 CA 证书包。
6. Kafka 表创建错误：如果表创建失败：
   • 验证仅使用本地 Kafka 连接器（强制要求）。
   • 检查 Kafka 主题是否存在且可访问。
   • 确保引导服务器配置正确。
   • 使用 validate_kafka_connector() 检查连接器的有效性。
7. 虚拟表不可查询：创建 Kafka 表后：
   • 重要提示：创建数据源并不会自动使其可查询。
   • 需要通过 SSB 界面手动在 Flink 目录中注册数据源。
   • 使用 SHOW TABLES; 查看哪些表实际上可以通过 SQL 查询。
   • 只有出现在 SHOW TABLES; 中的表才能通过 SQL 查询。

调试模式

通过设置环境变量启用调试日志：

export MCP_LOG_LEVEL=DEBUG

安全

• 所有敏感数据（密码、令牌、机密信息）在响应中会自动进行编辑。
• 大型集合会被截断，以防止使 LLM 过载。
• CDP 部署默认启用只读模式，以防止意外修改。
• 直接 SSB 认证使用基于 HTTP 的基本认证（适用于本地开发）。
• SQL 查询会自动进行清理，并正确终止分号。
• Kafka 表创建强制使用本地 Kafka 连接器，以确保数据安全。

总结

SSB MCP 服务器现在是一个全面的 SQL Stream Builder 管理平台，通过 80 多种 MCP 工具，使 Claude Desktop 能够访问几乎所有的 SSB 功能。

🎯 你将获得：

• 完整的 SSB 控制：管理作业、表、用户、项目和环境。
• 高级监控：系统健康、诊断和性能跟踪。
• 实时数据：灵活的采样和流数据访问。
• 企业功能：API 密钥、同步、导出/导入和多环境支持。
• 开发工具：UDF 管理、SQL 分析和连接器详细信息。
• DevOps 集成：项目管理、配置同步和 Git 工作流。

🚀 主要优势：

• 80% 以上的 API 覆盖率：几乎可以访问所有的 SSB 功能。
• 80 多种 MCP 工具：适用于各种用例的全面工具集。
• 15 个功能类别：功能组织有序，易于发现。
• 企业就绪：具备安全、监控和可扩展性等特性。
• 用户友好：通过 Claude Desktop 实现自然语言交互。
• 灵活性：支持独立部署和 CDP 部署。

📈 适用人群：

• 数据工程师：流处理、作业管理、实时分析。
• DevOps 团队：环境管理、监控、配置同步。
• 数据科学家：自定义 UDF、查询分析、数据探索。
• 平台管理员：用户管理、系统监控、安全。

SSB MCP 服务器将 Claude Desktop 转变为一个强大的 SSB 管理界面，支持与整个 SQL Stream Builder 环境进行自然语言交互！🎉

📄 许可证

本项目采用 Apache License 2.0 许可协议。