适合谁
- 需要与 Confluent Cloud 或 Kafka 交互的 AI 助手
- 使用 Confluent Platform 或本地 Kafka 的开发环境
- 需要统一工具集来管理 Kafka、Flink、Schema Registry 的团队
C
Confluent MCP Server
官方
让 AI 助手通过自然语言与 Confluent Cloud、Confluent Platform 及 Apache Kafka 交互的开源 MCP 服务器。
GitHub 源仓库 ↗
54FMRS · D
Confluent 官方开源 MCP 服务器，功能丰富，覆盖 Kafka、Flink、Schema Registry 等，但配置较复杂，且社区支持有限。
Confluent MCP Server 是一个开源服务器，提供 50+ 工具，覆盖 Kafka、Flink SQL、Schema Registry、Connectors、Tableflow 等，帮助 AI 助手管理 Kafka 集群、执行 Flink SQL、管理 schema 和连接器。适用于 Confluent Cloud、Confluent Platform 和本地 Kafka 部署。支持多种客户端，如 Claude、Cursor、VS Code 等。
工具能力
search-product-docs
搜索 Confluent 产品文档。
get-product-doc-page
获取 Confluent 产品文档页面的完整内容。
explain-disabled-tools
解释工具为何缺失。
list-configured-connections
列出所有已配置的连接及可用工具。
config-help
提供启用特定工具的配置建议。
describe-configured-connection
描述一个连接的配置和工具可用性。
list-topics
列出 Kafka 集群中的所有主题。
create-topics
创建一个或多个 Kafka 主题，可指定分区数和副本因子。
delete-topics
删除指定名称的 Kafka 主题。
produce-message
向 Kafka 主题生成消息，支持 Confluent Schema Registry 序列化。
consume-messages
从 Kafka 主题消费消息，可限制分区、偏移等。
list-consumer-groups
列出 Kafka 集群中的消费者组。
describe-consumer-group
描述单个消费者组的状态、类型、协议等。
get-consumer-group-lag
计算消费者组的偏移滞后。
alter-topic-config
修改 Kafka 主题配置。
get-topic-config
获取 Kafka 主题配置详情。
get-partition-offsets
返回主题各分区的低/高水位和消息数。
create-flink-statement
创建 Flink SQL 语句。
list-flink-statements
列出 Flink SQL 语句。
get-flink-statement-results
获取 Flink SQL 语句的结果行。
delete-flink-statements
删除 Flink SQL 语句。
get-flink-statement-exceptions
获取 Flink SQL 语句的最近异常。
list-compute-pools
列出 Flink 计算池。
list-flink-catalogs
列出 Flink 目录。
list-flink-databases
列出 Flink 数据库。
list-flink-tables
列出 Flink 表。
describe-flink-table
获取 Flink 表的完整 schema 详情。
get-flink-table-info
获取 Flink 表的元数据。
check-flink-statement-health
执行 Flink SQL 语句的健康检查。
detect-flink-statement-issues
检测 Flink SQL 语句的问题。
get-flink-statement-profile
获取 Flink SQL 语句的查询分析数据。
list-connectors
列出活跃的连接器。
get-connector-config
获取连接器的完整配置。
get-connector-offsets
获取连接器任务的当前偏移。
get-connector-status
获取连接器及其任务的状态。
get-connector-tasks
列出连接器的任务及其配置。
get-connector-error-summary
汇总连接器的当前错误。
get-connector-error-recommendations
获取连接器失败的补救建议。
get-connector-logs
获取连接器的近期日志。
create-connector
创建新连接器。
delete-connector
删除现有连接器。
pause-connector
暂停运行中的连接器及其任务。
resume-connector
恢复暂停的连接器及其任务。
restart-connector
重启连接器及其任务。
update-connector-config
更新现有连接器的配置。
list-schemas
列出 Schema Registry 中的所有 schemas。
create-schema
在 Schema Registry 中注册新 schema 或新版本。
delete-schema
从 Schema Registry 删除 schema subject 或特定版本。
search-topics-by-tag
按标签列出 Kafka 集群中的主题。
search-topics-by-name
按名称列出 Kafka 集群中的主题。
create-topic-tags
在 Confluent Cloud 中创建新的标签定义。
delete-tag
从 Confluent Cloud 删除标签定义。
remove-tag-from-entity
从实体中移除标签。
add-tags-to-topic
将现有标签分配给 Kafka 主题。
list-tags
列出所有标签及其定义。
list-organizations
列出当前凭据可见的 Confluent Cloud 组织。
list-environments
列出 Confluent Cloud 环境。
read-environment
按 ID 获取特定环境的详情。
list-clusters
获取 Confluent Cloud 环境中的所有集群。
create-tableflow-topic
创建 Tableflow 主题。
list-tableflow-topics
列出 Tableflow 主题。
read-tableflow-topic
读取 Tableflow 主题。
update-tableflow-topic
更新 Tableflow 主题。
delete-tableflow-topic
删除 Tableflow 主题。
list-tableflow-regions
列出 Tableflow 区域。
create-tableflow-catalog-integration
创建目录集成。
list-tableflow-catalog-integrations
列出目录集成。
read-tableflow-catalog-integration
读取目录集成。
update-tableflow-catalog-integration
更新目录集成。
delete-tableflow-catalog-integration
删除目录集成。
list-available-metrics
列出可用的 Confluent Cloud 指标。
query-metrics
查询 Confluent Cloud 指标。
list-billing-costs
查询账单和成本数据。
安装接入
- 确保 Node.js 22.19.0 或更高版本。2. 使用
npx @confluentinc/mcp-confluent --init-config生成配置。3. 编辑 config.yaml 文件。4. 使用npx @confluentinc/mcp-confluent --config ./config.yaml启动服务器。5. 在 MCP 客户端中配置连接。
选型与风险
不适合谁
- 非 Confluent 生态的 Kafka 用户（但支持通用 Kafka，仅部分工具可用）
- 希望完全托管 MCP 服务器的用户（可考虑 Confluent 官方托管服务）
- 需要社区支持以外官方支持的用户（社区支持仅为尽力而为）
所需权限
- 访问 Kafka 集群的主题、消费者组、配置
- 执行 Flink SQL 语句并读取结果
- 管理 Schema Registry 中的 schemas
- 管理 Connectors 和 Tableflow 主题
- 读取 Confluent Cloud 指标和账单数据
风险与副作用
- 工具可执行破坏性操作（如删除主题、删除 schema），需谨慎授权
- OAuth 登录凭据可能被泄露，需保护本地配置
- 使用 HTTP/SSE 传输时，若未设置 API 密钥，可能受到未授权访问
- Tableflow 工具需要云 IAM 权限，配置不当可能导致权限错误
常见排障
- Node.js 版本过低：升级到 22.19.0 以上
- 工具未出现：运行 `--list-tools` 检查工具是否启用，或调用 `explain-disabled-tools`
- 认证错误：使用 `--generate-key` 生成 API 密钥，并配置在 `server.auth.api_key`
- 端口冲突：在 `config.yaml` 中修改 `server.http.port`
- Tableflow 权限错误：按文档配置 IAM 角色和策略
使用场景
管理 Kafka 主题、消费者组和消息
执行 Flink SQL 查询并获取结果
管理 Schema Registry、连接器和 Tableflow
查询 Confluent Cloud 指标和账单信息
支持客户端
Claude Desktop完整支持
Claude Code完整支持
Cursor完整支持
VS Code完整支持
Goose完整支持
Gemini CLI完整支持