mcp-confluent

一个用于与 Confluent Kafka 和 Confluent Cloud REST API 交互的 MCP 服务器实现。

演示

Goose CLI

Claude Desktop

目录

• mcp-confluent
  • 演示
    • Goose CLI
    • Claude Desktop
  • 目录
  • 用户指南
    • 入门
    • 配置
    • 使用
    • 配置 Claude Desktop
    • 配置 Goose CLI
  • 开发者指南
    • 项目结构
    • 构建和运行
    • 测试
      • MCP Inspector
    • 添加新工具
    • 生成类型
    • 贡献

用户指南

入门

1. 创建 .env 文件： 将示例 .env 文件结构（如下所示）复制到项目根目录下的一个名为 .env 的新文件中。

2. 填充 .env 文件： 填写 Confluent Cloud 环境所需的必要值。有关每个变量的详细信息，请参阅 配置 部分。

3. 安装 Node.js（如果尚未安装）

   • 我们建议使用 NVM (Node Version Manager) 来管理 Node.js 版本
   • 安装并使用 Node.js：

   nvm install 22
   nvm use 22
   

配置

# .env 文件
BOOTSTRAP_SERVERS="pkc-v12gj.us-east4.gcp.confluent.cloud:9092"
KAFKA_API_KEY="..."
KAFKA_API_SECRET="..."
KAFKA_REST_ENDPOINT="https://pkc-v12gj.us-east4.gcp.confluent.cloud:443"
KAFKA_CLUSTER_ID=""
KAFKA_ENV_ID="env-..."
FLINK_ENV_ID="env-..."
FLINK_ORG_ID=""
FLINK_REST_ENDPOINT="https://flink.us-east4.gcp.confluent.cloud"
FLINK_ENV_NAME=""
FLINK_DATABASE_NAME=""
FLINK_API_KEY=""
FLINK_API_SECRET=""
FLINK_COMPUTE_POOL_ID="lfcp-..."
CONFLUENT_CLOUD_API_KEY=""
CONFLUENT_CLOUD_API_SECRET=""
CONFLUENT_CLOUD_REST_ENDPOINT="https://api.confluent.cloud"
SCHEMA_REGISTRY_API_KEY="..."
SCHEMA_REGISTRY_API_SECRET="..."
SCHEMA_REGISTRY_ENDPOINT="https://psrc-zv01y.northamerica-northeast2.gcp.confluent.cloud"

变量	描述	默认值	必需
BOOTSTRAP_SERVERS	Kafka broker 地址列表，格式为 host1:port1,host2:port2，用于建立与 Kafka 集群的初始连接 (字符串)		是
CONFIG_PATH	用于存储和检索基于会话的配置的文件系统路径，用于会话持久性（未来实现）(字符串)		是
CONFLUENT_CLOUD_API_KEY	用于 Confluent Cloud 平台管理的 Master API 密钥，允许管理整个组织的资源 (字符串 (最小长度: 1))		是
CONFLUENT_CLOUD_API_SECRET	与 CONFLUENT_CLOUD_API_KEY 配对的 Master API 密钥，用于全面的 Confluent Cloud 平台管理 (字符串 (最小长度: 1))		是
FLINK_API_KEY	用于访问 Confluent Cloud 的 Flink 服务的身份验证密钥，包括计算池和 SQL 语句管理 (字符串 (最小长度: 1))		是
FLINK_API_SECRET	与 FLINK_API_KEY 配对的密钥令牌，用于对 Confluent Cloud 的 Flink 服务进行身份验证访问 (字符串 (最小长度: 1))		是
KAFKA_API_KEY	建立与 Kafka 集群的安全连接所需的身份验证凭据（用户名）(字符串 (最小长度: 1))		是
KAFKA_API_SECRET	与 KAFKA_API_KEY 配对的身份验证凭据（密码），用于安全 Kafka 集群访问 (字符串 (最小长度: 1))		是
SCHEMA_REGISTRY_API_KEY	用于访问 Schema Registry 服务以管理和验证数据模式的身份验证密钥 (字符串 (最小长度: 1))		是
SCHEMA_REGISTRY_API_SECRET	与 SCHEMA_REGISTRY_API_KEY 配对的身份验证密钥，用于安全 Schema Registry 访问 (字符串 (最小长度: 1))		是
CONFLUENT_CLOUD_REST_ENDPOINT	Confluent Cloud 的 REST API 服务的基本 URL (默认)		否
FLINK_COMPUTE_POOL_ID	Flink 计算池的唯一标识符，必须以 'lfcp-' 前缀开头 (字符串)		否
FLINK_DATABASE_NAME	与 Kafka 集群关联的名称，用作 Flink SQL 操作中的数据库引用 (字符串 (最小长度: 1))		否
FLINK_ENV_ID	Flink 环境的唯一标识符，必须以 'env-' 前缀开头 (字符串)		否
FLINK_ENV_NAME	Flink 环境的人工可读名称，用于识别和显示目的 (字符串 (最小长度: 1))		否
FLINK_ORG_ID	Confluent Cloud 中用于 Flink 资源管理的组织标识符 (字符串 (最小长度: 1))		否
FLINK_REST_ENDPOINT	Confluent Cloud 的 Flink REST API 端点的基本 URL，用于 SQL 语句和计算池管理 (字符串)		否
KAFKA_CLUSTER_ID	Confluent Cloud 生态系统中 Kafka 集群的唯一标识符 (字符串 (最小长度: 1))		否
KAFKA_ENV_ID	Kafka 集群的环境标识符，必须以 'env-' 前缀开头 (字符串)		否
KAFKA_REST_ENDPOINT	用于 Kafka 集群管理和管理的 REST API 端点 (字符串)		否
SCHEMA_REGISTRY_ENDPOINT	用于访问 Schema Registry 服务以管理数据模式的 URL 端点 (字符串)		否

使用

此 MCP 服务器旨在与各种 MCP 客户端一起使用，例如 Claude Desktop 或 Goose CLI/Desktop。 具体配置和交互将取决于您使用的客户端。 但是，一般步骤是：

1. 构建： 按照 开发者指南 中的说明从源代码构建和运行服务器。 这通常涉及：

   • 安装依赖项 (npm install)
   • 构建项目 (npm run build 或 npm run dev)

2. 配置您的 MCP 客户端： 每个客户端都有其自己的方式来指定 MCP 服务器的地址和任何所需的凭据。 您需要配置您的客户端（例如，Claude、Goose）以连接到此服务器正在运行的地址（可能是 localhost 以及特定端口）。 服务器运行的端口可以通过环境变量配置。

3. 启动 MCP 客户端： 一旦您的客户端配置为连接到 MCP 服务器，您就可以启动您的 mcp 客户端，并且在启动时 - 它将在本地启动此 MCP 服务器的实例。 此实例将负责管理数据模式并代表您与 Confluent Cloud 交互。

4. 通过客户端与 Confluent 交互： 客户端连接后，您可以使用客户端的界面与 Confluent Cloud 资源交互。 客户端将向此 MCP 服务器发送请求，然后该服务器将代表您与 Confluent Cloud 交互。

配置 Claude Desktop

有关安装 claude desktop、mcp 服务器的更多详细信息，请参见 此处。

要配置 Claude Desktop 以使用此 MCP 服务器：

1. 打开 Claude Desktop 配置

   • 在 Mac 上：~/Library/Application Support/Claude/claude_desktop_config.json
   • 在 Windows 上：%APPDATA%\Claude\claude_desktop_config.json

2. 编辑配置文件

   • 在您喜欢的文本编辑器中打开配置文件
   • 添加或修改配置以包含 MCP 服务器：

   {
     "mcpServers": {
       "confluent": {
         "command": "node",
         "args": [
           "--env-file",
           "/path/to/confluent-mcp-server/.env",
           "/path/to/confluent-mcp-server/dist/index.js"
         ]
       }
     }
   }
   

   将 /path/to/confluent-mcp-server/ 替换为您安装此 MCP 服务器的实际路径。

3. 重启 Claude Desktop

   • 关闭并重新打开 Claude Desktop 以使更改生效
   • MCP 服务器将在 Claude Desktop 启动时自动启动

现在，Claude Desktop 将配置为使用您的本地 MCP 服务器进行 Confluent 交互。

配置 Goose CLI

有关如何安装 Goose CLI 的详细说明，请参见 此处。 安装完成后，您可以将此服务器添加为扩展

安装完成后，运行配置命令 goose configure 并选择以下内容

Add extension>Command-line Extension>mcp-confluent>node --env-file /path/to/confluent-mcp-server/.env /path/to/confluent-mcp-server/dist/index.js>No

将 /path/to/confluent-mcp-server/ 替换为您安装此 MCP 服务器的实际路径。

开发者指南

项目结构

/
├── src/                 # 源代码
│   ├── confluent/       # 与 Confluent 集成相关的代码（API 客户端等）
│   ├── tools/           # 工具实现（每个工具在一个单独的文件中）
│   ├── index.ts         # 服务器的主要入口点
│   └── ...              # 其他服务器逻辑、实用程序等
├── dist/                # 编译后的输出 (TypeScript -> JavaScript)
├── openapi.json         # Confluent Cloud 的 OpenAPI 规范
├── .env                 # 环境变量（示例 - 应复制并填写）
├── README.md            # 此文件
└── package.json         # Node.js 项目元数据和依赖项

构建和运行

1. 安装依赖项：

   npm install
   

2. 开发模式（监视更改）：

   npm run dev
   

   此命令将 TypeScript 代码编译为 JavaScript，并在 src/ 目录中检测到更改时自动重新构建。

3. 生产构建（一次性编译）：

   npm run build
   

4. 启动服务器：

   npm run start
   

测试

MCP Inspector

对于测试 MCP 服务器，您可以使用 MCP Inspector，它是一个交互式开发人员工具，用于测试和调试 MCP 服务器。

# 确保您已经以开发模式构建了项目，或者通过运行 npm run build
npx @modelcontextprotocol/inspector node --env-file $PATH_TO_PROJECT/.env /$PATH_TO_PROJECT/dist/index.js

添加新工具

1. 将新的枚举添加到枚举类 ToolName。
2. 将您的新工具添加到 ToolFactory 类中的处理程序映射。
3. 创建一个新文件，导出扩展 BaseToolHandler 的类。
   1. 实现基类的 handle 方法。
   2. 实现基类的 getToolConfig 方法。
4. 满意后，将其添加到 index.ts 中的 enabledTools 集合中。

生成类型

# 从 v7.5.2 开始，使用 allOf w/ required 时存在一个错误 https://github.com/openapi-ts/openapi-typescript/issues/1474。需要 --empty-objects-unknown 标志来避免它
npx openapi-typescript ./openapi.json -o ./src/confluent/openapi-schema.d.ts --empty-objects-unknown

贡献

<!-- 稍后填写。 -->