MCP Confluent

汇流MCP服务器

](https://www.npmjs.com/package/@confluentinc/mcp-confluent) 

开源 MCP服务器 它使AI助手能够通过自然语言与Confluent Cloud、Confluent Platform和独立的Apache Kafka部署进行交互。 它提供了50多种跨Kafka、Flink SQL、模式注册表、连接器、表流等工具，可用于任何兼容MCP的客户端，包括Claude Desktop、Claude Code、Cursor、VS Code、Goose和Gemini CLI。

快速开始

先决条件： . 如果你想与 汇流云，您需要先创建一个帐户。

1. 生成一个快速 config.yaml 项目根目录中的文件：

npx @confluentinc/mcp-confluent --init-config

1. 编辑 config.yaml 将您的连接详细信息归档，然后：

npx @confluentinc/mcp-confluent --config ./config.yaml

看 入门指南 有关完整的设置说明和 配置MCP客户端 与您首选的AI工具集成。

目录

• 快速开始
• 可用工具

- 始终可用 - 汇流云 - 本地部署

• 入门指南
• 配置
• 融合云的OAuth身份验证
• CLI使用情况
• 配置MCP客户端
• 遥测
• 故障排除
• 贡献

可用工具

工具会根据解析配置中存在的服务块自动启用；看见 配置.md 用于完整的块到工具映射。

您可以通过CLI列出所有可用工具：

npx -y @confluentinc/mcp-confluent --list-tools

始终可用的工具

这些工具不需要服务块或身份验证——即使在裸配置上也可以启用它们，而不管其余配置针对的是哪个部署。

Confluent Cloud的可用工具

这些工具需要端点和针对特定Confluent Cloud组件的身份验证。 参见 config.example.yaml 对于全套配置变量。 标有¹的类别也适用于 OAuth身份验证 -通过浏览器登录，而不是提供API密钥。

¹也可在OAuth下使用——请参阅 融合云的OAuth身份验证 有关设置和注意事项。 当前未标记的类别需要 direct 与静态API密钥的连接；OAuth迁移正在进行中。

用于本地部署的可用工具

这些工具只需要Kafka或Schema注册表端点-不需要Confluent Cloud API密钥/机密。 非常适合具有自我管理集群的本地发展，包括Confluent Platform。

# minimal config.yaml for local development
connections:
  local:
    type: direct
    kafka:
      bootstrap_servers: "localhost:9092"
    schema_registry:
      endpoint: "http://localhost:8081"

即用型变体已上线 sample_configs/.

入门指南

先决条件

• Node.js 22或更高版本 --我们建议使用 非易失性存储器 要管理版本，请执行以下操作：

  nvm install 22
  nvm use 22

• 运行Kafka或Schema Registry的本地环境，或 汇流云 具有适当API密钥或登录凭据的帐户，如果 使用OAuth进行身份验证.

一般设置步骤

此MCP服务器设计用于各种MCP客户端，如Claude Desktop、Copilot或Goose CLI/Desktop。 具体的配置和交互将取决于您使用的客户端。

MCP服务器可以通过以下方式向Confluent Cloud进行身份验证 OAuth（PKCE） 除了在YAML配置中定义的静态API密钥之外。 看 用于融合云的OAuth身份验证 了解更多详情。

配置（如果不使用OAuth）和运行此MCP的一般步骤是：

1. 创建配置文件： 复制提供的 config.yaml 例子 将文件保存到项目的根目录。

您可以使用CLI在当前目录中引导一个——不需要git checkout：

npx @confluentinc/mcp-confluent --init-config

1. 填充文件： 为您的Confluent Cloud环境填写必要的值。

看 配置.md 供完整参考；只填写您需要的服务块（每个服务块启用一组工具）。

1. 启动服务器： 您可以通过以下两种方式之一运行MCP服务器：

- 来源： 按照中的说明进行操作 贡献指南 从源代码构建和运行服务器。 这通常涉及：

- 安装依赖项(npm install) - 建设项目(npm run build 或 npm run dev)

- 使用npx： 您可以直接使用npx启动服务器，无需构建：

     npx @confluentinc/mcp-confluent --config /path/to/myconfig.yaml

1. 配置您的MCP客户端： 每个客户端（例如Claude、Goose）都有自己的方式指定MCP服务器的地址和任何所需的凭据。

你需要 配置您的客户端 连接到此服务器运行的地址（可能 localhost 具有特定端口）。 服务器运行的端口是通过设置的 server.http.port 在 config.yaml.

1. 启动您的MCP客户端： 一旦您的客户端配置为连接到MCP服务器，您就可以启动MCP客户端，启动时它将在本地建立此MCP服务器的实例。

此实例将负责管理数据模式并代表您与资源交互。

1. 通过客户与您的资源互动： 连接并配置客户端后，您可以使用客户端的界面与Confluent Cloud或本地资源进行交互。

客户端将向此MCP服务器发送请求，然后该服务器将代表您与可用连接进行交互。

配置

完整的配置参考——YAML模式、每个服务块、env-var插值、OAuth和HTTP/SSE认证设置、（已弃用的）遗留env-var表以及块到块映射工具——都存在于 配置.md.

兼容性说明。 此版本提供了YAML之间的完全对等(-c config.yaml)以及遗留的env var路径(-e config.env)对于单个连接。 env-var-only路径将在不久的将来的版本中发出启动警告，并在一两个版本后被删除。 多连接支持（下一版本）将仅支持YAML。 看 配置.md→ 两条路径，一种配置.

Tableflow命令的先决条件和设置

Tableflow工具通过Confluent cloud中的Flink运行时代表您与云存储（如AWS S3）和元数据目录（如AWS Glue）进行交互。 Flink运行时需要您的云帐户上的IAM权限，在任何Tableflow工具成功之前，必须授予这些权限并将其链接到Confluent cloud。

跟随 Tableflow快速启动，定制存储和胶水 设置角色、策略和提供程序集成。 当mcp-confluent尝试配置或管理启用了Tableflow的表时，跳过此步骤会导致授权错误。

融合云的OAuth身份验证

MCP服务器可以通过以下方式向Confluent Cloud进行身份验证 OAuth（PKCE） 而不是静态API密钥。 在需要Confluent访问的第一个工具调用中，服务器打开浏览器到Confluent Cloud登录页面；后续的工具调用重用生成的会话。 没有要提供的API密钥。

设置

npx @confluentinc/mcp-confluent --init-oauth-config
# edit ./config.yaml if needed, then:
npx @confluentinc/mcp-confluent --config ./config.yaml

--init-oauth-config 投掷起动器 config.oauth.example.yaml 进入 ./config.yaml. 整个文件基本上是：

connections:
  ccloud-oauth:
    type: oauth

看 配置.md→ 身份验证模式 为了获得完整的方案和人体工程学。

¹标记的类别 Confluent Cloud的可用工具 今天在OAuth下工作；其他一切都还需要 direct 与静态API密钥的连接。

CLI使用情况

MCP服务器为高级控制提供了灵活的命令行界面（CLI）。 CLI允许您选择配置文件、传输方式，并微调启用或阻止的工具。

基本用法

您可以查看所有CLI选项并获得以下帮助：

npx @confluentinc/mcp-confluent --help

Show output

Usage: mcp-confluent [options]

Confluent MCP Server - Model Context Protocol implementation for Confluent Cloud

Options:
  -V, --version                    output the version number
  -e, --env-file 
            Load environment variables from file
  -k, --kafka-config-file    Path to a properties file for configuring kafka clients
  -t, --transport           Transport types (comma-separated list) (choices: "http", "sse", "stdio", default: "stdio")
  --allow-tools             Comma-separated list of tool names to allow. If provided, takes precedence over --allow-tools-file. Allow-list is applied before block-list.
  --block-tools             Comma-separated list of tool names to block. If provided, takes precedence over --block-tools-file. Block-list is applied after allow-list.
  --allow-tools-file         File with tool names to allow (one per line). Used only if --allow-tools is not provided. Allow-list is applied before block-list.
  --block-tools-file         File with tool names to block (one per line). Used only if --block-tools is not provided. Block-list is applied after allow-list.
  --list-tools                     Print the final set of enabled tool names (with descriptions) after allow/block filtering and exit. Does not start the server.
  --disable-auth                   Disable authentication for HTTP/SSE transports. WARNING: Only use in development environments.
  --allowed-hosts           Comma-separated list of allowed Host header values for DNS rebinding protection.
  --generate-key                   Generate a secure API key for MCP_API_KEY and print it to stdout, then exit.
  -h, --help                       display help for command

示例：使用所有传输进行部署

npx @confluentinc/mcp-confluent -c config.yaml --transport http,sse,stdio

Show output

...
{"level":"info","time":"2025-05-14T17:03:02.883Z","pid":47959,"hostname":"G9PW1FJH64","name":"mcp-confluent","msg":"Starting transports: http, sse, stdio"}
{"level":"info","time":"2025-05-14T17:03:02.971Z","pid":47959,"hostname":"G9PW1FJH64","name":"mcp-confluent","msg":"HTTP transport routes registered"}
{"level":"info","time":"2025-05-14T17:03:02.972Z","pid":47959,"hostname":"G9PW1FJH64","name":"mcp-confluent","msg":"SSE transport routes registered"}
{"level":"info","time":"2025-05-14T17:03:02.972Z","pid":47959,"hostname":"G9PW1FJH64","name":"mcp-confluent","msg":"STDIO transport connected"}
{"level":"info","time":"2025-05-14T17:03:03.012Z","pid":47959,"hostname":"G9PW1FJH64","name":"mcp-confluent","msg":"Server listening at http://[::1]:3000"}
{"level":"info","time":"2025-05-14T17:03:03.013Z","pid":47959,"hostname":"G9PW1FJH64","name":"mcp-confluent","msg":"Server listening at http://127.0.0.1:3000"}
{"level":"info","time":"2025-05-14T17:03:03.013Z","pid":47959,"hostname":"G9PW1FJH64","name":"mcp-confluent","msg":"All transports started successfully"}

示例：只允许使用特定工具

npx @confluentinc/mcp-confluent -c config.yaml --allow-tools produce-message,consume-messages

仅启用指定的工具；所有其他人都将被禁用。

示例：阻止某些工具

npx @confluentinc/mcp-confluent -c config.yaml --block-tools produce-message,consume-messages

除指定工具外，所有工具都将启用。

示例：使用文件中的工具列表

您还可以在文件中维护允许/阻止列表（每行一个工具名称）：

npx -y @confluentinc/mcp-confluent -c config.yaml --allow-tools-file allow.txt --block-tools-file block.txt

示例：列出所有可用工具

npx -y @confluentinc/mcp-confluent --list-tools

Show output

add-tags-to-topic: Assign existing tags to Kafka topics in Confluent Cloud.
alter-topic-config: Alter topic configuration in Confluent Cloud.
consume-messages: Consumes messages from one or more Kafka topics. Supports automatic deserialization of Schema Registry encoded messag...
create-connector: Create a new connector. Returns the new connector information if successful.
create-flink-statement: Make a request to create a statement.
create-topic-tags: Create new tag definitions in Confluent Cloud.
create-topics: Create one or more Kafka topics.
delete-connector: Delete an existing connector. Returns success message if deletion was successful.
delete-flink-statements: Make a request to delete a statement.
delete-tag: Delete a tag definition from Confluent Cloud.
delete-topics: Delete the topic with the given names.
check-flink-statement-health: Perform an aggregate health check for a Flink SQL statement.
describe-flink-table: Get full schema details for a Flink table via INFORMATION_SCHEMA.COLUMNS.
detect-flink-statement-issues: Detect issues for a Flink SQL statement by analyzing status, exceptions, and metrics.
get-flink-statement-profile: Get Query Profiler data with task graph, metrics, and automated issue detection.
get-flink-table-info: Get table metadata via INFORMATION_SCHEMA.TABLES.
list-flink-catalogs: List all catalogs in the Flink environment.
list-flink-databases: List all databases (schemas) in a Flink catalog via INFORMATION_SCHEMA.SCHEMATA.
list-flink-tables: List all tables in a Flink database.
get-flink-statement-exceptions: Retrieve the 10 most recent exceptions for a Flink SQL statement.
get-topic-config: Retrieve configuration details for a specific Kafka topic.
list-clusters: Get all clusters in the Confluent Cloud environment
list-connectors: Retrieve a list of "names" of the active connectors. You can then make a read request for a specific connector by name.
list-environments: Get all environments in Confluent Cloud with pagination support
list-flink-statements: Retrieve a sorted, filtered, paginated list of all statements.
list-schemas: List all schemas in the Schema Registry.
list-tags: Retrieve all tags with definitions from Confluent Cloud Schema Registry.
list-topics: List all topics in the Kafka cluster.
produce-message: Produce records to a Kafka topic. Supports Confluent Schema Registry serialization (AVRO, JSON, PROTOBUF) for both ke...
read-connector: Get information about the connector.
read-environment: Get details of a specific environment by ID
read-flink-statement: Make a request to read a statement and its results
remove-tag-from-entity: Remove tag from an entity in Confluent Cloud.
search-topics-by-name: List all topics in the Kafka cluster matching the specified name.
search-topics-by-tag: List all topics in the Kafka cluster with the specified tag.
create-tableflow-topic: Make a request to create a tableflow topic.
list-tableflow-regions: Retrieve a sorted, filtered, paginated list of all tableflow regions.
list-tableflow-topics: Retrieve a sorted, filtered, paginated list of all tableflow topics.
read-tableflow-topic: Make a request to read a tableflow topic.
update-tableflow-topic: Make a request to update a tableflow topic.
delete-tableflow-topic: Make a request to delete a tableflow topic.
create-tableflow-catalog-integration: Make a request to create a catalog integration.
list-tableflow-catalog-integrations: Retrieve a sorted, filtered, paginated list of all catalog integrations.
read-tableflow-catalog-integration: Make a request to read a catalog integration.
update-tableflow-catalog-integration: Make a request to update a catalog integration.
delete-tableflow-catalog-integration: Make a request to delete a tableflow catalog integration.
list-organizations: List Confluent Cloud organizations the current credentials can see. Paginated; if the response includes a nextPageToken, pass it back as pageToken to fetch additional pages.
explain-disabled-tools: Call when the user asks why a tool is missing or unavailable (e.g., "why can't I list Kafka topics?", "where are the Flink tools?"). Returns disabled tools grouped by the config gap each one is waiting on, so you can tell the user the exact YAML block or field to add. Prefer this over guessing about credentials, network, or auth.

提示： 允许列表在阻止列表之前应用。 如果两者都没有提供，则默认启用所有工具。

配置MCP客户端

请参阅以下指南，了解如何在首选客户端上设置和使用此MCP服务器的分步说明：

• 克劳德代码
• 克劳德桌面
• 光标
• Gemini CLI
• Goose命令行界面
• IBM Bob
• VS Code
• 帆板运动

遥测

此MCP服务器收集使用数据以帮助改进。 您可以通过设置退出 DO_NOT_TRACK=true 在您的环境中。 看 telemetry.md 有关收集内容的完整详细信息。

故障排除

“不支持Node.js版本” --此项目需要Node.js 22或更高版本。 检查您的版本 node -v 并在需要时进行升级。

工具未出现 --每个工具都需要您的特定服务块 config.yaml. 跑 --list-tools 查看哪些工具处于活动状态，或调用 explain-disabled-tools 出于每个工具的原因，客户提供MCP工具。 块到工具映射存在于 配置.md.

HTTP/SSE上的身份验证错误 --使用生成API密钥 npx @confluentinc/mcp-confluent --generate-key 并将其添加到您的 config.yaml 在...之下 server.auth.api_key. 看 配置.md→ HTTP/SSE传输安全.

连接被拒绝/端口冲突 --默认的HTTP端口是8080。 集 server.http.port 在你的 config.yaml 改变它。

表流授权错误 --表流工具需要您的云环境中具有特定的IAM权限。 看 Tableflow命令的先决条件和设置.

贡献

Bug报告和反馈以Github Issues的形式表示。 有关投稿指南，请参阅 贡献.md

发布前测试

要针对预发布版本运行MCP服务器进行beta测试或早期反馈，请将发布tarball文件下载到本地目录。 然后，当运行任何 npx 上面的命令，替换 @confluentinc/mcp-confluent 指向那个tarball的路径，例如。 npx @~path/to/my/tarball --list-tools