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 認證（獨立部署）

Knox 認證（CDP）

* 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 許可協議。