Kiro 中配置 RabbitMQ MCP Server 指南一、背景介绍通过配置 RabbitMQ MCP Server可以让 Kiro 具备直接与 RabbitMQ 消息队列交互的能力包括发送消息、查看队列状态、消费消息等方便在开发过程中进行消息驱动的功能测试验证。二、方案选型方案优缺点开源 PyPI 包uvx 运行安装简单但依赖兼容性差版本更新可能破坏功能自建本地 Python MCP Server稳定可控仅依赖pikamcp两个库无第三方包兼容问题推荐方案自建本地 Python MCP Server注博客https://blog.csdn.net/badao_liumang_qizhi三、环境准备1. 确认 Python 环境python--version# 需要 Python 3.102. 安装依赖pipinstallpika mcppikaPython 的 RabbitMQ/AMQP 客户端库mcpModel Context Protocol SDK用于构建 MCP Server四、创建 MCP Server 脚本在项目的.kiro/tools/目录下创建rabbitmq_mcp_server.py RabbitMQ MCP Server - 本地版. 支持标准 RabbitMQ 及阿里云 ApsaraMQ for RabbitMQ。 importjsonimportosimportpikafrommcp.server.fastmcpimportFastMCP# 从环境变量获取连接配置RABBITMQ_HOSTos.environ.get(RABBITMQ_HOST,localhost)RABBITMQ_PORTint(os.environ.get(RABBITMQ_PORT,5672))RABBITMQ_USERNAMEos.environ.get(RABBITMQ_USERNAME,guest)RABBITMQ_PASSWORDos.environ.get(RABBITMQ_PASSWORD,guest)RABBITMQ_VHOSTos.environ.get(RABBITMQ_VHOST,/)mcpFastMCP(rabbitmq-server)defget_connection():创建RabbitMQ连接.credentialspika.PlainCredentials(RABBITMQ_USERNAME,RABBITMQ_PASSWORD)parameterspika.ConnectionParameters(hostRABBITMQ_HOST,portRABBITMQ_PORT,virtual_hostRABBITMQ_VHOST,credentialscredentials,connection_attempts3,retry_delay5,socket_timeout30,heartbeat0,)returnpika.BlockingConnection(parameters)mcp.tool()defpublish_message(exchange:str,routing_key:str,message:str,content_type:strapplication/json)-str:发送消息到指定的exchange和routing_key.try:connectionget_connection()channelconnection.channel()propertiespika.BasicProperties(content_typecontent_type,delivery_mode2,)channel.basic_publish(exchangeexchange,routing_keyrouting_key,bodymessage.encode(utf-8),propertiesproperties,)connection.close()returnf消息发送成功。exchange{exchange}, routing_key{routing_key}exceptExceptionase:returnf消息发送失败:{str(e)}mcp.tool()defcheck_queue(queue_name:str)-str:检查指定队列的状态.try:connectionget_connection()channelconnection.channel()resultchannel.queue_declare(queuequeue_name,passiveTrue)info{queue:result.method.queue,message_count:result.method.message_count,consumer_count:result.method.consumer_count,}connection.close()returnjson.dumps(info,ensure_asciiFalse)exceptExceptionase:returnf队列检查失败:{str(e)}mcp.tool()defget_message(queue_name:str,auto_ack:boolFalse)-str:从指定队列中获取一条消息.try:connectionget_connection()channelconnection.channel()method,properties,bodychannel.basic_get(queuequeue_name,auto_ackauto_ack)ifmethodisNone:connection.close()return队列中没有消息msg_info{delivery_tag:method.delivery_tag,exchange:method.exchange,routing_key:method.routing_key,content_type:properties.content_type,body:body.decode(utf-8)ifbodyelse,}ifnotauto_ack:channel.basic_nack(delivery_tagmethod.delivery_tag,requeueTrue)connection.close()returnjson.dumps(msg_info,ensure_asciiFalse,indent2)exceptExceptionase:returnf获取消息失败:{str(e)}mcp.tool()deftest_connection()-str:测试RabbitMQ连接是否正常.try:connectionget_connection()connection.channel()connection.close()returnf连接成功host{RABBITMQ_HOST}, port{RABBITMQ_PORT}, vhost{RABBITMQ_VHOST}exceptExceptionase:returnf连接失败:{str(e)}if__name____main__:mcp.run(transportstdio)五、配置 MCP Server在.kiro/settings/mcp.json中添加配置{mcpServers:{rabbitmq:{command:python,args:[脚本的绝对路径/rabbitmq_mcp_server.py],env:{RABBITMQ_HOST:your-rabbitmq-host.com,RABBITMQ_PORT:5672,RABBITMQ_USERNAME:your-username,RABBITMQ_PASSWORD:your-password,RABBITMQ_VHOST:/},disabled:false,autoApprove:[]}}}注意事项args中的脚本路径必须使用绝对路径因为 Kiro 的工作目录不是项目目录Windows 路径示例D:\\Project\\my-app\\.kiro\\tools\\rabbitmq_mcp_server.py六、阿里云 ApsaraMQ 适配说明如果使用阿里云 ApsaraMQ for RabbitMQ认证方式有所不同阿里云不支持直接使用 AccessKey 作为用户名密码需要先在阿里云控制台通过 AK/SK生成静态用户名和密码生成后的静态凭据可以直接配置在RABBITMQ_USERNAME和RABBITMQ_PASSWORD中协议层面完全兼容无需额外适配七、问题排查问题1脚本路径找不到python: cant open file .kiro/tools/rabbitmq_mcp_server.py: No such file or directory原因使用了相对路径Kiro 的工作目录不在项目根目录下解决改为绝对路径问题2PyPI 包依赖冲突ImportError: cannot import name BearerAuthProvider from fastmcp.server.auth原因开源 PyPI 包与 fastmcp 版本不兼容解决放弃第三方包使用自建本地脚本方案问题3国内网络无法访问 PyPIFailed to fetch: https://pypi.org/simple/xxx - operation timed out解决使用国内镜像或直接采用本地脚本方案无需网络下载问题4消费者反序列化失败Failed to invoke target method with argument type [class [B]原因Java Spring 项目使用SimpleMessageConverterJava 序列化Python 发送的 JSON 无法被正确反序列化解决对于 Java 序列化的消息通过 Java 单元测试发送消息Python MCP 仅用于查看队列状态和消费已有消息八、使用示例测试连接 测试下 RabbitMQ 连接 → 调用 test_connection() ← 连接成功hostyour-host.com, port5672, vhost/查看队列状态 查看 my-queue 队列的消息数 → 调用 check_queue(my-queue) ← {queue: my-queue, message_count: 3, consumer_count: 1}发送消息 发送一条测试消息到 my-exchange → 调用 publish_message(my-exchange, my-routing-key, {key: value}) ← 消息发送成功。exchangemy-exchange, routing_keymy-routing-key读取消息不删除 查看队列中的第一条消息 → 调用 get_message(my-queue, auto_ackFalse) ← {delivery_tag: 1, exchange: my-exchange, routing_key: ..., body: ...}九、适用场景与局限适用场景查看队列状态消息堆积数、消费者数发送简单的 JSON 文本消息读取/窥视队列中的消息内容验证消息路由是否正确局限无法发送 Java 序列化格式的消息需要通过 Java 单测配合不支持管理操作创建/删除 exchange、queue 等需 Management API不支持 TLS/SSL 连接如需要可在脚本中添加 SSL 配置十、总结本地自建 Python MCP Server 是最稳定可靠的方案配合 MySQL MCP 可以实现完整的消息驱动功能验证闭环MySQL MCP→ 查询/准备测试数据RabbitMQ MCP→ 查看队列状态、发送/读取消息MySQL MCP→ 验证消费后的数据库状态变化