Kafka MCP Server

Kafka MCP 服务

一个使用 Go 实现的 Apache Kafka 模型上下文协议（MCP）服务，基于 franz-go 和 mcp-go。

此服务提供了通过 MCP 协议与 Kafka 交互的实现，使大模型能够通过标准化接口执行常见的 Kafka 操作。

概述

Kafka MCP 服务弥合了大模型与 Apache Kafka 之间的差距，使其能够：

• 从主题发布和消费消息
• 列出、描述和管理主题
• 监控和管理消费者组
• 评估集群运行状况和配置
• 执行标准 Kafka 操作

所有操作均通过标准化的模型上下文协议（MCP）完成。

架构

graph TB
    subgraph "MCP Client (AI Applications)"
        A[Claude Desktop]
        B[Cursor]
        C[Windsurf]
        D[ChatWise]
    end
    
    subgraph "Kafka MCP Server"
        E[MCP Protocol Handler]
        F[Tools Registry]
        G[Resources Registry]
        H[Prompts Registry]
        I[Kafka Client Wrapper]
    end
    
    subgraph "Apache Kafka Cluster"
        J[Broker 1]
        K[Broker 2]
        L[Broker 3]
        M[Topics & Partitions]
        N[Consumer Groups]
    end
    
    A --> E
    B --> E
    C --> E
    D --> E
    
    E --> F
    E --> G
    E --> H
    
    F --> I
    G --> I
    H --> I
    
    I --> J
    I --> K
    I --> L
    
    J --> M
    K --> M
    L --> M
    
    J --> N
    K --> N
    L --> N
    
    classDef client fill:#e1f5fe
    classDef mcp fill:#f3e5f5
    classDef kafka fill:#fff3e0
    
    class A,B,C,D client
    class E,F,G,H,I mcp
    class J,K,L,M,N kafka

工作原理：

1. MCP 客户端（AI 应用）通过 stdio 或 HTTP 传输连接到 Kafka MCP 服务
2. MCP 服务提供三类功能：
   • 工具 - 直接执行 Kafka 操作（发布/消费消息、描述主题等）
   • 资源 - 集群运行状况报告和诊断信息
   • 提示词 - 为常见操作预配置的工作流
3. Kafka 客户端封装层使用 franz-go 库处理所有 Kafka 通信
4. Apache Kafka 集群处理实际的消息流和存储

传输模式：

• STDIO：默认模式，适用于本地 MCP 客户端（Claude Desktop、Cursor 等）
• HTTP：支持远程访问，可选择使用 OAuth 2.1 身份验证

主要功能

• Kafka 集成：通过 MCP 实现常见的 Kafka 操作
• 安全性：
  • 支持 SASL（PLAIN、SCRAM-SHA-256、SCRAM-SHA-512）和 TLS 身份验证
  • 支持 HTTP 传输的 OAuth 2.1 身份验证（Native 和 Proxy 模式）
  • 支持 Okta、Google、Azure AD 和 HMAC 提供方
• 灵活的传输方式：本地客户端使用 STDIO，远程访问使用 HTTP
• 错误处理：提供有意义的反馈
• 配置选项：可针对不同环境自定义
• 预配置提示词：为常见 Kafka 操作提供一组提示词
• 兼容性：适用于兼容 MCP 的大模型

开始使用

前提条件

• Go 1.24 或更高版本
• Docker（用于运行集成测试）
• 可访问 Kafka 集群

安装

Homebrew（macOS 和 Linux）

安装 kafka-mcp-server 最简单的方法是使用 Homebrew：

# Add the tap repository
brew tap tuannvm/mcp

# Install kafka-mcp-server
brew install kafka-mcp-server

要更新到最新版本：

brew update && brew upgrade kafka-mcp-server

从源码安装

# Clone the repository
git clone https://github.com/tuannvm/kafka-mcp-server.git
cd kafka-mcp-server

# Build the server
go build -o kafka-mcp-server ./cmd

MCP 客户端集成

此 MCP 服务可以集成到多种 AI 应用中。以下是针对不同平台的说明：

Cursor

编辑 ~/.cursor/mcp.json，并添加 kafka-mcp-server 配置：

{
  "mcpServers": {
    "kafka": {
      "command": "kafka-mcp-server",
      "args": [],
      "env": {
        "KAFKA_BROKERS": "localhost:9092",
        "KAFKA_CLIENT_ID": "kafka-mcp-server",
        "MCP_TRANSPORT": "stdio"
      }
    }
  }
}

Claude Desktop

编辑 Claude 配置文件，并添加服务：

• macOS：~/Library/Application Support/Claude/claude_desktop_config.json
• Windows：%APPDATA%\Claude\claude_desktop_config.json

{
  "mcpServers": {
    "kafka": {
      "command": "kafka-mcp-server",
      "args": [],
      "env": {
        "KAFKA_BROKERS": "localhost:9092",
        "KAFKA_CLIENT_ID": "kafka-mcp-server",
        "MCP_TRANSPORT": "stdio"
      }
    }
  }
}

重启 Claude Desktop 以应用更改。

Claude Code

要与 Claude Code 配合使用，请使用内置的 MCP 配置命令添加服务：

# Add kafka-mcp-server with environment variables
claude mcp add kafka \
  --env KAFKA_BROKERS=localhost:9092 \
  --env KAFKA_CLIENT_ID=kafka-mcp-server \
  --env MCP_TRANSPORT=stdio \
  --env KAFKA_SASL_MECHANISM= \
  --env KAFKA_SASL_USER= \
  --env KAFKA_SASL_PASSWORD= \
  --env KAFKA_TLS_ENABLE=false \
  -- kafka-mcp-server

其他实用命令：

# List configured MCP servers
claude mcp list

# Remove server
claude mcp remove kafka

# Test server connection
claude mcp get kafka

ChatWise

1. 打开 ChatWise → 设置 → 工具 → “+” → “命令行 MCP”
2. 配置：
   • ID：kafka
   • 命令：kafka-mcp-server
   • 参数：（留空）
   • 环境变量：添加环境变量：

     KAFKA_BROKERS=localhost:9092
     KAFKA_CLIENT_ID=kafka-mcp-server
     MCP_TRANSPORT=stdio
     

使用 mcpenetes 简化配置

在多个客户端之间管理 MCP 服务配置可能会变得很棘手。mcpenetes 是一款专用工具，可大幅简化这一流程：

# Install mcpenetes
go install github.com/tuannvm/mcpenetes@latest

主要功能

• 交互式搜索：通过简单的命令查找并选择 Kafka MCP 服务配置
• 应用到所有客户端：自动在所有 MCP 客户端之间同步配置
• 配置备份：在进行更改前安全地备份现有配置
• 恢复：需要时轻松还原到之前的配置

mcpenetes 快速入门

# Search for available MCP servers including kafka-mcp-server
mcpenetes search 

# Apply kafka-mcp-server configuration to all your clients at once
mcpenetes apply

# Load a configuration from your clipboard
mcpenetes load

使用 mcpenetes，你可以维护多个 Kafka 配置（开发、生产等），并在所有客户端（Cursor、Claude Desktop、Windsurf、ChatWise）之间即时切换，而无需手动编辑每个客户端的配置文件。

MCP 工具

此服务提供以下用于与 Kafka 交互的工具。如需包含示例和响应样例的详细文档，请参阅 docs/tools.md。

• produce_message：向 Kafka 主题生成消息
• consume_messages：通过批量操作从 Kafka 主题消费消息
• list_brokers：列出所有已配置的 Kafka Broker 地址
• describe_topic：提供特定主题的完整元数据
• list_consumer_groups：列出集群中的所有消费者组
• describe_consumer_group：提供详细的消费者组信息，包括滞后指标
• describe_configs：检索 Kafka 资源的配置设置
• cluster_overview：提供全面的集群运行状况摘要
• list_topics：列出所有主题及其元数据，包括分区和副本信息

MCP 资源

此服务提供以下可通过 MCP 协议访问的资源。如需包含响应示例的详细文档，请参阅 docs/resources.md。

• kafka-mcp://overview：全面的集群运行状况摘要
• kafka-mcp://health-check：详细的运行状况评估及可执行的建议
• kafka-mcp://under-replicated-partitions：分析存在副本问题的分区
• kafka-mcp://consumer-lag-report：可自定义阈值的消费者性能分析

MCP 提示词

该服务包含以下针对 Kafka 运维和诊断的预配置提示词。有关参数和示例响应的详细文档，请参阅 docs/prompts.md。

• kafka_cluster_overview：生成全面的集群运行状况摘要
• kafka_health_check：执行详细的运行状况评估，并提供可操作的建议
• kafka_under_replicated_partitions：分析副本问题并提供故障排查指导
• kafka_consumer_lag_report：分析消费者延迟，并提供性能优化建议

端到端示例

以下交互示例展示了 AI 助手如何使用这些工具帮助用户：

用户查询：“我们需要调查订单处理流水线为何落后。你能帮我检查一下消费者延迟吗？”

AI 助手的工作流：

1. 首先，检查消费者组

   > Using list_consumer_groups tool
   > Discovers "order-processor" group
   

2. 然后，查看消费者组详情

   > Using describe_consumer_group tool with group_id "order-processor" and include_offsets=true
   > Discovers lag on several partitions
   

3. 获取集群概览以检查总体运行状况

   > Using cluster_overview tool
   > Confirms no under-replicated partitions or offline brokers
   

4. 生成消费者延迟报告

   > Using consumer_lag_report prompt with threshold=1000
   > Provides detailed analysis of lag across consumer groups
   

5. 提供可操作的建议：

   Based on my analysis, your order-processor consumer group has significant lag on topic orders, partitions 2 and 3, with lag values of 15,420 and 12,305 messages respectively.
   
   The cluster itself is healthy with no under-replicated partitions, suggesting this is an issue with the consumer application rather than Kafka infrastructure.
   
   I recommend:
   1. Check the consumer instances processing those partitions for slowdowns or errors
   2. Consider scaling up your consumer group by adding more instances
   3. Review the consumer configuration, particularly max.poll.records and max.poll.interval.ms
   4. Examine any processing bottlenecks in the consumer logic
   

这一无缝工作流展示了 Kafka MCP 工具如何让大模型执行复杂的诊断并提供可行的见解。

配置

可使用以下环境变量配置该服务：

变量	描述	默认值
KAFKA_BROKERS	以逗号分隔的 Kafka broker 地址列表	localhost:9092
KAFKA_CLIENT_ID	用于连接的 Kafka 客户端 ID	kafka-mcp-server
MCP_TRANSPORT	MCP 传输方式（stdio/http）	stdio
KAFKA_SASL_MECHANISM	SASL 机制：plain、scram-sha-256、scram-sha-512 或 ""（禁用）	""
KAFKA_SASL_USER	SASL 身份验证用户名	""
KAFKA_SASL_PASSWORD	SASL 身份验证密码	""
KAFKA_TLS_ENABLE	为 Kafka 连接启用 TLS（true 或 false）	false
KAFKA_TLS_INSECURE_SKIP_VERIFY	跳过 TLS 证书验证（true 或 false）	false

OAuth 2.1 配置（仅限 HTTP 传输）

使用 HTTP 传输（MCP_TRANSPORT=http）时，可以启用 OAuth 2.1 身份验证：

变量	描述	默认值	必需
MCP_HTTP_PORT	HTTP 服务端口	8080	否
OAUTH_ENABLED	启用 OAuth 2.1 身份验证	false	否
OAUTH_MODE	OAuth 模式：native 或 proxy	native	否
OAUTH_PROVIDER	提供方：hmac、okta、google、azure	okta	否
OAUTH_SERVER_URL	完整的 MCP 服务 URL（例如，https://localhost:8080）	-	启用 OAuth 时必需
OIDC_ISSUER	OAuth 颁发者 URL	-	启用 OAuth 时必需
OIDC_AUDIENCE	OAuth 受众	-	启用 OAuth 时必需
OIDC_CLIENT_ID	OAuth 客户端 ID	-	仅代理模式
OIDC_CLIENT_SECRET	OAuth 客户端密钥	-	仅代理模式
OAUTH_REDIRECT_URIS	以逗号分隔的重定向 URI	-	仅代理模式
JWT_SECRET	JWT 签名密钥	-	仅代理模式

有关 OAuth 设置和示例的详细信息，请参阅 docs/oauth.md。

安全说明：

• 使用 KAFKA_TLS_INSECURE_SKIP_VERIFY=true 时，该服务将跳过 TLS 证书验证。此设置仅应在开发或测试环境中使用，或在使用自签名证书时使用。
• 只有使用 HTTP 传输时才能使用 OAuth。STDIO 传输不支持 OAuth。
• 启用 OAuth 后，生产环境中务必使用 HTTPS。

安全注意事项

该服务在设计时考虑了企业级安全性：

• 身份验证：
  • Kafka：全面支持 SASL PLAIN、SCRAM-SHA-256 和 SCRAM-SHA-512
  • MCP Server：HTTP 传输使用 OAuth 2.1 身份验证（Okta、Google、Azure AD、HMAC）
• 加密：支持 TLS，确保与 Kafka broker 安全通信
• 输入验证：全面验证所有用户输入，以防止注入攻击
• 错误处理：安全地处理错误，不泄露敏感信息
• 令牌安全：验证 OAuth 保护端点的 Bearer 令牌，并缓存 5 分钟

OAuth 安全最佳实践请参阅 docs/oauth.md。

开发

测试

全面的测试覆盖确保了可靠性：

# Run all tests (requires Docker for integration tests)
go test ./...

# Run tests excluding integration tests
go test -short ./...

# Run integration tests with specific Kafka brokers
export KAFKA_BROKERS="your-broker:9092"
export SKIP_KAFKA_TESTS="false"
go test ./kafka -v -run Test

贡献

欢迎贡献！请随时提交 Pull Request。

许可证

本项目采用 MIT License 授权，详情请参阅 LICENSE 文件。