MCP Kafka Client

Kafka MCP服务器

用于与Apache Kafka交互的MCP（模型上下文协议）服务器，包括支持Avro的Schema Registry。

🚀 特性

• 主题列表：获取Kafka集群中的所有可用主题
• 架构查询：从架构注册表读取Avro架构
• 有效载荷生成：根据Avro模式生成有效的测试有效载荷
• 消息生成：通过自动Avro序列化向Kafka发送消息
• 消息消费：阅读和分析来自Avro反序列化主题的消息
• 模式分析：详细分析Avro模式的结构

📋 需求

• Python 3.10或更高版本
• 访问Kafka集群
• 冲突模式注册表

🔧 安装

1. 克隆或创建项目

cd kafka-mcp-server

1. 创建虚拟环境

python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

1. 安装依赖项

pip install -r requirements.txt

1. 配置环境变量

复制示例文件并使用您的凭据进行编辑：

cp .env.example .env

编辑 .env 文件：

# Kafka Configuration
KAFKA_BOOTSTRAP_SERVERS=your-kafka-broker.example.com:9092

# Schema Registry Configuration
SCHEMA_REGISTRY_URL=https://your-schema-registry.example.com

# Kafka Security
KAFKA_SECURITY_PROTOCOL=SASL_SSL
KAFKA_SASL_MECHANISM=OAUTHBEARER

🎯 用法

运行MCP服务器

python -m kafka_mcp.server

在Claude桌面中配置

编辑您的Claude Desktop配置文件(claude_desktop_config.json):

{
  "mcpServers": {
    "kafka": {
      "command": "python",
      "args": ["-m", "kafka_mcp.server"],
      "cwd": "/full/path/to/kafka-mcp-server",
      "env": {
        "KAFKA_BOOTSTRAP_SERVERS": "your-kafka-broker.example.com:9092",
        "AZURE_TENANT_ID": "your-tenant-id",
        "AZURE_CLIENT_ID": "your-client-id",
        "AZURE_CLIENT_SECRET": "your-client-secret",
        "AZURE_SCOPE": "https://your-kafka-resource/.default",
        "SCHEMA_REGISTRY_URL": "https://your-schema-registry.example.com",
        "KAFKA_SECURITY_PROTOCOL": "SASL_SSL",
        "KAFKA_SASL_MECHANISM": "OAUTHBEARER"
      }
    }
  }
}

🛠️ 可用工具

1. list_topics

列出Kafka集群中的所有可用主题。

参数：

• timeout （可选）：超时时间（秒）（默认值：10.0）

Claude中的使用示例：

List all available Kafka topics

2. get_topic_schema

从架构注册表获取主题的Avro架构。

参数：

• topic （必填）：主题名称
• is_key （可选）：如果为true，则获取密钥模式；如果没有，则使用值模式（默认值：false）

Claude中的使用示例：

Get the schema for topic "users"

3. generate_payload

根据主题的Avro模式生成有效的测试有效负载。

参数：

• topic （必填）：主题名称
• custom_values （可选）：特定字段的自定义值
• count （可选）：要生成的有效载荷数（默认值：1）
• is_key （可选）：如果为true，则为键模式生成（默认值：false）

Claude中的使用示例：

Generate 5 test payloads for topic "orders"

使用自定义值：

Generate a payload for topic "users" with email "test@example.com" and name "John Doe"

4. produce_message

通过自动Avro序列化向Kafka发送消息。

参数：

• topic （必填）：主题名称
• value （必填）：消息值为JSON对象
• key （可选）：消息键为 字符串 或 JSON对象

- If string：编码为UTF-8字节 - 如果JSON对象：使用Avro模式（如果可用）或JSON序列化

• schema （可选）：JSON格式的Avro模式（如果未提供，则从注册表中获取）
• key_schema （可选）：密钥的Avro模式

Claude中的使用示例：

使用字符串键：

Send a message to topic "users" with key "user-123" and payload: {"id": "123", "name": "Alice", "email": "alice@example.com"}

使用JSON对象键：

Send a message to topic "users" with key {"userId": "123"} and payload: {"name": "Alice", "email": "alice@example.com"}

无钥匙：

Send a message to topic "users" with this payload: {"id": "123", "name": "Alice", "email": "alice@example.com"}

5. consume_messages

使用Avro反序列化读取和分析主题中的消息。

参数：

• topic （必填）：主题名称
• num_messages （可选）：要使用的消息数（默认值：10）
• from_beginning （可选）：如果为true，则从主题的开头读取（默认值：false）
• group_id （可选）：消费者组ID。如果未提供，则 自动生成唯一组ID 对于每个请求，避免并行消费者之间的冲突

Claude中的使用示例：

Read the last 20 messages from topic "orders"

注： 默认情况下，每个消费请求都使用一个唯一的消费者组，允许多个并行读取而不会发生冲突。您可以指定自定义 group_id 如果您需要跨请求跟踪消费者偏移。

6. analyze_schema

分析Avro模式并提供有关其结构的详细信息。

参数：

• topic （必填）：主题名称
• is_key （可选）：如果为true，则分析密钥模式（默认值：false）

Claude中的使用示例：

Analyze the schema for topic "products" and show me all fields

身份验证流程

1. 服务器使用Azure凭据获取访问令牌
2. 令牌作为SASL/OAUTHBEARER包含在每个Kafka连接中
3. 令牌到期时会自动续订

📊 智能有效载荷生成

有效载荷生成器包括基于字段名称的智能逻辑：

• email：生成有效的电子邮件（例如。， user123@example.com)
• name：生成常用名称
• id/uuid：生成数字ID
• url/uri：生成有效的URL
• phone：生成电话号码
• date/timestamp：生成ISO日期/时间戳
• status：生成常见状态（活动、非活动、待定、已完成）
• country：生成国家代码（美国、英国、欧洲等）
• currency：生成货币代码（美元、欧元、英镑等）

🧪 完整使用示例

典型测试工作流程

# 1. List available topics
"List all topics"

# 2. View topic schema
"Show me the schema for topic 'user-events'"

# 3. Analyze schema structure
"Analyze the schema for topic 'user-events' and explain each field"

# 4. Generate test payloads
"Generate 3 test payloads for 'user-events'"

# 5. Send test message
"Send a test message to topic 'user-events' with userId='test-123'"

# 6. Verify it was sent
"Read the last 5 messages from 'user-events'"

消息调试

# Read recent messages
"Read the last 50 messages from topic 'error-logs'"

# Analyze patterns
"Analyze messages from topic 'transactions' and look for patterns in the last 100 messages"

📁 项目结构

kafka-mcp-server/
├── src/
│   └── kafka_mcp/
│       ├── __init__.py
│       ├── __main__.py          # Entry point
│       ├── server.py             # Main MCP server
│       ├── config.py             # Configuration
│       ├── auth.py               # Azure OAuth authentication
│       ├── kafka_client.py       # Kafka and Schema Registry client
│       └── payload_generator.py  # Avro payload generator
├── tests/                        # Tests (to be implemented)
├── .env.example                  # Configuration example
├── .gitignore
├── pyproject.toml
├── requirements.txt
└── README.md

🐛 故障排除

错误：“获取OAuth令牌失败”

• 验证您的Azure凭据是否正确
• 检查一下 AZURE_SCOPE 是正确的
• 检查应用程序是否具有Kafka资源的权限

错误：“找不到主题的架构”

• 验证该主题是否存在并且具有已注册的架构
• 检查架构注册表URL
• 确保主题遵循格式 -value 或 -key

错误：“连接超时”

• 验证与Kafka集群的网络连接
• 检查一下 KAFKA_BOOTSTRAP_SERVERS 是正确的
• 检查SSL/TLS配置（如适用）

🔜 路线图

• \[\]单元和集成测试
• \[\]支持JSON模式和Protobuf
• \[\]指标和监测
• \[\]用于快速测试的独立CLI
• \[\]支持消息转换
• \[\]Kafka Connect集成

📝 许可证

MIT许可证

🤝 贡献

欢迎捐款。拜托：

1. 分叉项目
2. 为您的功能创建分支(git checkout -b feature/AmazingFeature)
3. 提交您的更改(git commit -m 'Add some AmazingFeature')
4. 推到分支(git push origin feature/AmazingFeature)
5. 打开拉取请求

👥 作者

耶稣罗德里格斯

🙏 致谢

• Kafka库的融合平台
• MCP协议的拟人化
• Apache Kafka社区