使用说明

WSO2 MI+Kafka MCP服务器

一个本地开发平台，允许您直接从AI代理控制Apache Kafka和WSO2 Micro Integrator。一个提示设置整个堆栈。26个MCP工具处理之后的一切——主题管理、消息发布、工件生成、诊断等等。

适用于 克劳德桌面版, 克劳德代码, 光标, 帆板运动, 法典, 泽德, Continue.dev，以及任何兼容MCP的客户端。

核心成果： 你克隆这个仓库，在你的AI客户端注册它，比如说 “设置卡夫卡和米”，并在Docker中运行一个完全工作的Kafka+WSO2 MI集成栈——无需手动Docker命令，无需YAML编辑，无需配置步骤。

所得

运行中的Kafka代理、ZooKeeper、Kafka UI和WSO2微集成器——都在Docker中
一个可用的演示流程：HTTP POST→ 卡夫卡→ WSO2 MI消费者→ 丰富审计主题
死信队列（DLQ）的错误处理
26个MCP工具，用于管理主题、发布/使用消息、生成MI工件、跟踪消息流和故障排除
一切都通过克劳德的自然语言控制

这是给谁的

集成工程师 在本地构建Kafka+WSO2 MI流
开发者 谁想要一个单命令本地Kafka/MI堆栈进行测试
人工智能辅助开发人员 使用任何兼容MCP的客户端：Claude Desktop、Claude Code、Cursor、Windsurf、Codex、Zed、Continue.dev或其他

快速开始

这将使您从一个新的克隆到一个正在运行的堆栈。按顺序执行每一步。

步骤1：检查先决条件

你需要Docker和Node.js。运行以下命令进行验证：

docker --version        # Need Docker with Compose v2
docker compose version  # Should print "Docker Compose version v2.x.x"
node --version          # Need v18 or higher

如果 docker compose version 失败，安装或更新Docker Desktop、Rancher Desktop或Colima。

步骤2：克隆和构建

git clone https://github.com/PasinduGunarathne/wso2-mi-kafka-mcp-server.git
cd wso2-mi-kafka-mcp-server
npm install
npm run build

这 npm run build 步骤创建 dist/ 目录。没有它，MCP服务器将无法工作。

步骤3：获取绝对路径

下一步你需要这个。复制以下输出：

echo "$(pwd)/dist/index.js"

输出示例： /Users/you/wso2-mi-kafka-mcp-server/dist/index.js

步骤4：注册MCP服务器

此服务器使用 stdio传输 --通用MCP标准。它适用于任何兼容MCP的AI客户端。在下面选择您的客户。

在以下所有示例中，替换 /absolute/path/to 使用您在步骤3中获得的实际路径。

克劳德桌面版

打开（或创建）配置文件：

macOS： ~/Library/Application Support/Claude/claude_desktop_config.json
窗户： %APPDATA%\Claude\claude_desktop_config.json

{
  "mcpServers": {
    "kafka-mi": {
      "command": "node",
      "args": ["/absolute/path/to/wso2-mi-kafka-mcp-server/dist/index.js"]
    }
  }
}

重新启动： 完全退出并重新打开Claude Desktop（而不仅仅是关闭窗口）。 验证： 在新对话中查找锤子/工具图标→ 应该看看 setup_kafka_and_mi.

克劳德代码（CLI）

claude mcp add kafka-mi node /absolute/path/to/wso2-mi-kafka-mcp-server/dist/index.js

验证： claude mcp list → 应该显示 kafka-mi.

光标

打开 设置→ MCP （或 Cmd/Ctrl+Shift+P → “MCP:添加服务器”），然后添加到您的MCP配置中：

{
  "mcpServers": {
    "kafka-mi": {
      "command": "node",
      "args": ["/absolute/path/to/wso2-mi-kafka-mcp-server/dist/index.js"]
    }
  }
}

重新启动： 重新启动游标或重新加载窗口。 验证： 打开MCP面板→ kafka-mi 应显示为与26个工具连接。

风帆冲浪（Codeium）

打开 设置→ MCP服务器 （或 Cmd/Ctrl+Shift+P → “MCP：配置”），然后添加：

{
  "mcpServers": {
    "kafka-mi": {
      "command": "node",
      "args": ["/absolute/path/to/wso2-mi-kafka-mcp-server/dist/index.js"]
    }
  }
}

重新启动： 重新加载Windsurf窗口。 验证： MCP服务器面板应显示 kafka-mi 作为活跃。

Codex（OpenAI CLI）

添加到您的 ~/.codex/config.json （或项目 .codex/config.json):

{
  "mcpServers": {
    "kafka-mi": {
      "command": "node",
      "args": ["/absolute/path/to/wso2-mi-kafka-mcp-server/dist/index.js"]
    }
  }
}

然后在启用MCP的情况下运行Codex：

codex --model o4-mini

验证： Codex将在启动时列出可用的MCP工具。

泽德

打开设置 (Cmd+,)并添加到您的 settings.json 在...之下 "context_servers":

{
  "context_servers": {
    "kafka-mi": {
      "command": {
        "path": "node",
        "args": ["/absolute/path/to/wso2-mi-kafka-mcp-server/dist/index.js"]
      },
      "settings": {}
    }
  }
}

重新启动： 重新加载Zed。 验证： 打开“助手”面板→ 服务器工具应该可用。

Continue.dev

添加到Continue配置文件(~/.continue/config.yaml 或 .continue/config.yaml):

mcpServers:
  - name: kafka-mi
    command: node
    args:
      - /absolute/path/to/wso2-mi-kafka-mcp-server/dist/index.js

重新启动： 重新加载Continue扩展。 验证： MCP工具应出现在继续工具面板中。

任何其他MCP客户端

此服务器与 支持MCP stdio传输的任何客户端通用配置模式为：

字段	值
命令	`node`
参数	`["/absolute/path/to/wso2-mi-kafka-mcp-server/dist/index.js"]`
运输	stdio（默认）

服务器公开了26个工具。一旦连接，说 "setup kafka and mi" 开始。

步骤5：运行第一个设置

确保Docker正在运行，然后打开你的AI客户端并键入：

setup kafka and mi

您的AI代理将提供两个选项：

选项A:Docker Hub镜像 --提取WSO2 MI的官方图像。不需要本地文件。最适合大多数用户。
选项B：本地MI包 --使用a wso2mi-<version>.zip 你提供。用于自定义构建或企业包。

选择一个（例如“使用Docker Hub镜像”或“使用本地MI包”）。然后，代理运行完整的设置，第一次运行需要3-5分钟：

检查Docker和Node.js是否可用
生成中的所有项目文件 ~/kafka-mi-demo
构建Docker镜像（下载WSO2 MI、Kafka连接器JAR）
启动4个容器（ZooKeeper、Kafka、Kafka UI、WSO2 MI）
创建3个Kafka主题
运行烟雾测试以验证流量是否正常

步骤6：确认它有效

设置完成后，告诉您的AI代理：

run health checks

所有检查都应通过。您也可以手动验证：

检查	如何
集装箱正在运行	`docker ps` --应显示4个集装箱 `demo-` 前缀
Kafka UI可访问	打开http://localhost:8090在浏览器中
WSO2 MI可访问	打开http://localhost:8290/kafka/health--应该返回 `{"status":"ok"}`
主题已创建	在Kafka UI中，单击“主题”-应该看到 `demo.orders.in`, `demo.orders.audit`, `demo.orders.dlq`

你的前10分钟

堆栈运行后，在AI客户端中尝试以下提示：

1.运行演示

run demo

这将通过完整流程发布一个示例订单事件：HTTP→ 卡夫卡→ WSO2 MI消费者→ 审计主题。克劳德在每一步都给你看证据。

2.测试错误处理

trigger error

这会发送一条故意使处理失败的消息，并被路由到死信队列。

3.检查DLQ

check dlq

显示带有错误详细信息和原始有效载荷的失败消息。

4.探索卡夫卡

list kafka topics

consume 5 messages from demo.orders.audit

成功是什么样子的

完成这些步骤后，您应该：

4个Docker容器运行良好(docker ps)
Kafka UI中可见的3个Kafka主题http://localhost:8090
在中可见的消息 demo.orders.in （生）， demo.orders.audit （丰富），以及 demo.orders.dlq （失败）
WSO2 MI在http://localhost:8290/kafka/health

建筑

以下是消息在系统中流动时发生的情况：

HTTP Client ──POST /kafka/publish──> WSO2 MI ──kafkaTransport──> demo.orders.in (Kafka)
                                                                        │
                                     WSO2 MI <──KafkaOrderConsumer──────┘
                                        │
                                   ┌────┴────┐
                                Success    Error
                                   │         │
                          demo.orders.audit  demo.orders.dlq
                          (enriched events)  (dead letter queue)

用简单的英语：

您（或Claude）通过HTTP POST向WSO2 MI发送JSON订单事件
WSO2 MI验证它并将其发布到 demo.orders.in 卡夫卡主题
WSO2 MI中的Kafka入站端点轮询该主题并消费消息
处理序列对事件类型进行验证、丰富和分支
关于成功：丰富的信息是 demo.orders.audit
出错时：消息转到 demo.orders.dlq 附有错误详细信息

服务和港口

服务	端口	URL	目的
卡夫卡经纪人	9092	`localhost:9092`	消息代理
Kafka用户界面	8090	http://localhost:8090	用于浏览主题和消息的Web UI
WSO2 MI HTTP	8290	http://localhost:8290	集成API端点
WSO2 MI HTTPS	8253	https://localhost:8253	安全的API端点
WSO2 MI管理	9164	http://localhost:9164	MI管理/管理员API

卡夫卡主题

主题	目的
`demo.orders.in`	传入订单事件（通过HTTP API发布）
`demo.orders.audit`	丰富的事件（消费、处理、转发）
`demo.orders.dlq`	死信队列（失败/无效消息）

按目的列出的常见提示

将这些复制粘贴到您的AI客户端中，以使用MCP工具。

堆栈管理：

setup kafka and mi              (shows option menu first)
get mi config
stack status
show logs for wso2mi
start stack
stop stack

演示和测试：

run demo
run health checks
trigger error with type missing-id
check dlq

Kafka主题管理：

list kafka topics
describe topic demo.orders.in
create a topic called payments.events with 5 partitions
delete topic payments.events

发布和消费消息：

publish {"id":"2001","customer":"Bob","event":"order-created","amount":50} to demo.orders.in
consume 10 messages from demo.orders.audit
search demo.orders.in for customer Alice
replay DLQ messages back to input

MI人工制品脚手架：

generate an API called PaymentAPI with context /payments
generate a sequence called paymentProcessingSequence
generate a Kafka inbound endpoint for topic payments.events
validate my MI artifacts

诊断：

trace order demo-1234 through the flow
collect stack diagnostics
smoke test POST http://localhost:8290/kafka/publish with body {"id":"test","event":"order-created","customer":"Test","amount":1}

手动测试（无Claude）

如果你想直接从终端测试正在运行的堆栈，请使用curl：

curl -X POST http://localhost:8290/kafka/publish \
  -H "Content-Type: application/json" \
  -d '{"id":"1001","customer":"Alice","event":"order-created","amount":125.50}'

预期响应：

{"status":"published","topic":"demo.orders.in","orderId":"1001","event":"order-created","timestamp":"...","message":"Event successfully published to Kafka"}

这对于验证堆栈是否独立于Claude工作或脚本测试非常有用。

完整的MCP工具参考（26）

堆栈管理

工具	说明
`setup_kafka_and_mi`	全自动设置——先决条件、构建、启动、主题、冒烟测试。支持Docker Hub或本地MI包。
`start_stack`	启动所有容器
`stop_stack`	停止所有容器（保留数据）
`stack_status`	显示容器运行状况、主题、API可用性
`show_logs`	查看最近的日志（所有或每个服务：wso2mi、kafka、zookeeper、kafka-ui）
`reset_environment`	停止+删除所有容器和卷（需要确认=true）
`get_mi_config`	显示当前MI源（Docker Hub或本地包）、版本和可用选项

演示与测试

工具	说明
`run_demo`	发布示例事件并验证端到端的完整流程
`run_health_checks`	验证所有服务是否正常
`trigger_error`	测试错误处理--路由到死信队列（类型：处理、缺少id、缺少事件）
`check_dlq`	查看DLQ中包含错误详细信息的失败消息

Kafka管理

工具	说明
`list_kafka_topics`	列出所有主题（不包括内部主题）
`describe_kafka_topic`	显示主题的分区、复制、配置、ISR
`create_kafka_topic`	使用自定义分区和复制因子创建新主题
`delete_kafka_topic`	删除主题（需要confirm=true）

消息操作

工具	说明
`publish_message`	将消息直接发布到任何Kafka主题（支持可选键）
`consume_messages`	使用分区、偏移量和时间戳详细信息消费消息
`search_messages`	按子字符串搜索主题中的消息（不区分大小写）
`replay_dlq_to_input`	将DLQ消息回放到输入主题进行重新处理（需要confirm=true）

MI人工制品脚手架

工具	说明
`generate_mi_api`	从模板生成WSO2 MI API工件
`generate_mi_sequence`	生成带有日志记录和错误处理的中介序列
`generate_mi_inbound_endpoint`	生成Kafka入站端点消费者
`validate_mi_artifacts`	验证Synapse XML工件是否存在常见问题

诊断和可追溯性

工具	说明
`trace_order_flow`	在整个流程中跟踪订单（输入→ 消费者→ 审计/DLQ）
`collect_stack_diagnostics`	全面的诊断——容器、主题、API、错误、磁盘使用情况
`smoke_test_custom_flow`	使用可选的Kafka主题验证测试自定义HTTP端点

故障排除

Docker未运行

Error: Cannot connect to the Docker daemon

在运行安装程序之前，请启动Docker Desktop、Rancher Desktop或Colima。

端口已在使用中

Error: port is already allocated

另一项服务正在使用所需的端口之一（8090、8290、8253、9092、9164）。停止冲突的服务或更改中的端口 docker-compose.yml.

AI客户端看不到MCP服务器

你建造了吗？ 跑 npm run build --the dist/index.js 文件必须存在。
路径正确吗？ 配置中的路径必须是绝对的（以开头 / 在macOS/Linux或 C:\ 在Windows上），而不是相对的。
你重启了吗？ 大多数客户端在配置更改后需要重新启动或重新加载窗口。
客户特定检查：
- 克劳德桌面： 完全退出并重新打开（而不仅仅是关闭窗口）。查找工具图标。
- 克劳德代码： 跑 claude mcp list 以验证注册。
- 光标： 设置→ MCP → check kafka-mi 连接状态。
- 风浪： MCP服务器面板→ check kafka-mi 是活跃的。
- 食品法典： 检查一下 ~/.codex/config.json 有正确的路径。
- Zed： 检查 settings.json → context_servers 部分。
- Continue.dev： 检查 config.yaml → mcpServers 部分。

容器已启动，但尚不健康

WSO2 MI需要60-90秒才能完全启动。如果安装后健康检查立即失败，请等待一分钟，然后重试：

run health checks

未找到主题

如果topic命令失败，Kafka容器可能没有准备好。请检查：

stack status

如果Kafka显示为不健康，请等待它完成启动或检查日志：

show logs for kafka

有些东西坏了，你想重新开始

reset environment

这将停止所有容器并删除所有数据卷。然后运行 setup kafka and mi 再一次。

日常工作流程

开始新的一天（堆栈已停止）

start stack

然后等待约60秒，让WSO2 MI恢复健康。验证 run health checks.

完成后停止

stop stack

这将保留所有数据。当你重新开始时，你的主题和消息仍然存在。

检查正在运行的内容

stack status

查看日志

show logs                     # all services
show logs for wso2mi          # just WSO2 MI
show logs for kafka           # just Kafka

完全重置（删除所有内容）

reset environment

然后 setup kafka and mi 重新开始。

高级：MI图像选项

大多数用户可以跳过此部分。默认的Docker Hub镜像可以开箱即用。

这 setup_kafka_and_mi 该工具允许您通过以下方式在两个MI源之间进行选择 miSource 参数：

选项1:Docker Hub镜像（默认）

不需要本地文件。告诉克劳德：

setup kafka and mi

或者使用特定版本：

setup kafka and mi with MI version 4.5.0

可用版本： 4.3.0, 4.4.0, 4.5.0 （默认）。追加 -alpine 或 -rocky 对于变体（例如。 4.5.0-alpine).

选项2：本地MI分发ZIP

从下载WSO2 MI分发ZIP WSO2
将其放置在MCP服务器项目根目录中： wso2mi-kafka-mcp-server/wso2mi-4.4.0.zip
告诉克劳德：

setup kafka and mi using local MI pack

该工具会自动检测ZIP并提取版本。如果存在多个ZIP，请明确指定版本：

setup kafka and mi using local MI pack version 4.5.0

检查当前配置

get mi config

显示活动源代码、版本、检测到的本地ZIP以及所有可用的Docker Hub版本。

手动构建（无MCP）

# Docker Hub (default)
MI_VERSION=4.5.0 docker compose build wso2mi

# Local pack
MI_DOCKERFILE=Dockerfile MI_VERSION=4.5.0 docker compose build wso2mi

你的选择是坚持的 ~/kafka-mi-demo/.env 以便 start_stack 并且所有组合命令都自动使用相同的设置。

高级：可扩展性

对于想要构建自定义集成流的高级用户：

人工脚手架： 使用 generate_mi_api, generate_mi_sequence，以及 generate_mi_inbound_endpoint 从模板创建新的Synapse XML工件。这些生成了具有适当名称空间、日志记录和错误处理的生产就绪样板。
验证： 使用 validate_mi_artifacts 在部署之前检查XML文件是否存在常见错误（缺少属性、名称空间错误、Kafka配置问题）。
自定义烟雾测试： 使用 smoke_test_custom_flow 测试任何HTTP端点，并可选地验证消息是否出现在Kafka主题中。
订单跟踪： 使用 trace_order_flow 在管道的每个阶段（输入主题、MI消费者日志、审计主题和DLQ）遵循特定的订单ID。

项目结构

wso2-mi-kafka-mcp-server/
├── src/
│   ├── index.ts                    # Server entry point — registers all 26 MCP tools
│   ├── types.ts                    # TypeScript interfaces for topics, messages, traces
│   ├── tools/
│   │   ├── setup.ts                # One-command stack setup (setup_kafka_and_mi)
│   │   ├── demo.ts                 # Demo flow, error triggers, DLQ checks, health checks
│   │   ├── stack.ts                # Start, stop, status, logs, reset
│   │   ├── kafka-admin.ts          # Topic CRUD operations
│   │   ├── kafka-messages.ts       # Publish, consume, search, replay messages
│   │   ├── mi-artifacts.ts         # Generate and validate WSO2 MI XML artifacts
│   │   └── diagnostics.ts          # Order tracing, diagnostics, custom smoke tests
│   ├── services/
│   │   ├── kafka-service.ts        # Runs kafka-* CLI commands inside the Kafka container
│   │   └── mi-service.ts           # Calls WSO2 MI management API, renders XML templates
│   └── utils/
│       ├── config.ts               # MI source config persistence (.env read/write)
│       ├── docker.ts               # Docker and Compose command wrappers with timeouts
│       ├── files.ts                # Copies resource files into the project directory
│       ├── logger.ts               # Colored output formatting for tool results
│       └── validation.ts           # Input validation (topic names, required fields)
├── resources/
│   ├── artifacts/                  # Pre-built WSO2 MI Synapse XML artifacts
│   │   ├── apis/                   #   HTTP API that publishes to Kafka
│   │   ├── sequences/              #   Processing sequence + fault/DLQ sequence
│   │   ├── inbound-endpoints/      #   Kafka consumer endpoint
│   │   ├── local-entries/          #   Kafka producer connection config
│   │   └── imports/                #   Kafka connector import declaration
│   ├── templates/                  # Mustache templates for artifact generation
│   ├── conf/deployment.toml        # WSO2 MI server configuration
│   ├── docker/                     # Dockerfiles and docker-compose.yml
│   └── scripts/                    # Shell scripts for topic creation and testing
├── claude-desktop-config.json      # Example MCP config (copy into your Claude config)
├── package.json
└── tsconfig.json

MI Docker镜像依赖关系

dropin/（OSGi捆绑包）： mi-inbound-kafka-2.0.6、kafka-avro序列化器-7.6.0.wso2v1、kafka-模式序列化器-7-6.0.wso2v1、kafka模式调节客户端-7.6.0.wso2v1、org.apache.avro-1.11.3（自定义OSGi包）

lib/（类路径JAR）： kafka-clients-3.6.1、通用配置-7.6.0、通用工具-7.6.0和度量标准-2.2.0

突触库/（连接器）： kafkaTransport连接器-3.2.0.zip

参考：https://mi.docs.wso2.com/en/latest/reference/connectors/kafka-connector/setting-up-kafka/