1. 为什么要在 Kiro 里接一个 RabbitMQ MCP Server如果你正在用 Kiro 做本地开发联调大概率遇到过这种场景代码里发了一条消息但不确定到底有没有进队列、堆积了多少、消费者有没有正常消费。以前的做法是切到 RabbitMQ 管理后台或者写一段临时脚本手动basic_get来回折腾。Kiro 本身支持 MCPModel Context Protocol把 RabbitMQ 封装成一个 MCP Server 之后你就能直接在对话里让它帮你查队列、发消息、读消息联调效率会明显不一样。这篇要解决的就是这件事在 Kiro 里配置一个本地自建的 RabbitMQ MCP Server用 TaoToken 统一 Key 走 API 通道把settings.jsonKiro 里实际是mcp.json骨架一次填对启动后能验证连通、能收发消息。适合谁适合本地跑 RabbitMQ、用 Kiro 做 AI 辅助编码、想快速验证消息驱动逻辑的开发者。Python 3.10 环境即可不需要额外装一堆第三方包。我试过直接用 PyPI 上的开源 MCP 包结果版本一升级就报ImportError后来改成自建本地脚本只依赖pika和mcp两个库稳定很多。下面按“环境准备 → 脚本 → 配置 → 验证 → 排障”的顺序走一遍。2. TaoToken 前置统一 Key 与 API 通道怎么填TaoToken 在这里的角色是给 Kiro 提供统一的模型调用入口。Kiro 的 Agent 能力需要模型支撑而 MCP Server 负责工具执行两者配合才能完成“对话 → 调工具 → 返回结果”的闭环。所以你需要先在 TaoToken 侧拿到 Key再把它填进 Kiro 的配置里。具体动作第一打开 TaoToken 控制台创建 API Key。地址是https://taotoken.net/api-keys登录后新建一个 Key复制保存。这个 Key 就是后面配置里要填的凭证。第二确认 API 通道地址。TaoToken 的 API 入口是https://taotoken.net/api注意这个地址不带任何查询参数直接作为 base URL 使用。第三如果你打算长期用 Kiro 做编码和 Agent 任务建议看一下 Coding Plan地址https://taotoken.net/coding-plan它针对高频编码场景做了额度优化比按量调用更划算。只是想先验证模型对话的话用模型对话页https://taotoken.net/models快速试一下也行。注意Key 只创建一次就够MCP Server 本身不消耗模型额度消耗额度的是 Kiro 里的模型调用。两者分开理解配置时不要混。3. 可复制配置settings.json 骨架与 MCP Server 脚本Kiro 的 MCP 配置文件名是mcp.json放在项目.kiro/settings/目录下。很多人搜“settings.json 骨架”其实指的就是这个文件的结构。下面给出完整骨架你直接改路径和连接信息即可。3.1 环境准备先确认 Python 版本必须是 3.10 及以上python --version # 需要 Python 3.10然后安装两个依赖pip install pika mcppika是 RabbitMQ 的 AMQP 客户端mcp是 Model Context Protocol 的 SDK用来把普通函数暴露成 MCP 工具。就这两个没有其他第三方包避免了版本冲突。3.2 创建 MCP Server 脚本在项目.kiro/tools/目录下新建rabbitmq_mcp_server.py。核心逻辑是从环境变量读连接配置用FastMCP注册四个工具——测试连接、发消息、查队列、读消息。RabbitMQ MCP Server - 本地版支持标准 RabbitMQ。 import json import os import pika from mcp.server.fastmcp import FastMCP RABBITMQ_HOST os.environ.get(RABBITMQ_HOST, localhost) RABBITMQ_PORT int(os.environ.get(RABBITMQ_PORT, 5672)) RABBITMQ_USERNAME os.environ.get(RABBITMQ_USERNAME, guest) RABBITMQ_PASSWORD os.environ.get(RABBITMQ_PASSWORD, guest) RABBITMQ_VHOST os.environ.get(RABBITMQ_VHOST, /) mcp FastMCP(rabbitmq-server) def get_connection(): credentials pika.PlainCredentials(RABBITMQ_USERNAME, RABBITMQ_PASSWORD) parameters pika.ConnectionParameters( hostRABBITMQ_HOST, portRABBITMQ_PORT, virtual_hostRABBITMQ_VHOST, credentialscredentials, connection_attempts3, retry_delay5, socket_timeout30, heartbeat0, ) return pika.BlockingConnection(parameters) mcp.tool() def test_connection() - str: 测试 RabbitMQ 连接是否正常。 try: connection get_connection() connection.channel() connection.close() return f连接成功host{RABBITMQ_HOST}, port{RABBITMQ_PORT}, vhost{RABBITMQ_VHOST} except Exception as e: return f连接失败: {str(e)} mcp.tool() def publish_message(exchange: str, routing_key: str, message: str, content_type: str application/json) - str: 发送消息到指定的 exchange 和 routing_key。 try: connection get_connection() channel connection.channel() properties pika.BasicProperties(content_typecontent_type, delivery_mode2) channel.basic_publish( exchangeexchange, routing_keyrouting_key, bodymessage.encode(utf-8), propertiesproperties, ) connection.close() return f消息发送成功。exchange{exchange}, routing_key{routing_key} except Exception as e: return f消息发送失败: {str(e)} mcp.tool() def check_queue(queue_name: str) - str: 检查指定队列的状态。 try: connection get_connection() channel connection.channel() result channel.queue_declare(queuequeue_name, passiveTrue) info { queue: result.method.queue, message_count: result.method.message_count, consumer_count: result.method.consumer_count, } connection.close() return json.dumps(info, ensure_asciiFalse) except Exception as e: return f队列检查失败: {str(e)} mcp.tool() def get_message(queue_name: str, auto_ack: bool False) - str: 从指定队列中获取一条消息默认不删除。 try: connection get_connection() channel connection.channel() method, properties, body channel.basic_get(queuequeue_name, auto_ackauto_ack) if method is None: 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) if body else , } if not auto_ack: channel.basic_nack(delivery_tagmethod.delivery_tag, requeueTrue) connection.close() return json.dumps(msg_info, ensure_asciiFalse, indent2) except Exception as e: return f获取消息失败: {str(e)} if __name__ __main__: mcp.run(transportstdio)这里有个细节值得说get_message默认auto_ackFalse读完会basic_nack并requeueTrue相当于“窥视”消息而不消费掉联调时特别有用不会因为读一次就把消息弄丢。3.3 mcp.json 骨架在.kiro/settings/mcp.json里写入{ mcpServers: { rabbitmq: { command: python, args: [D:\\Project\\my-app\\.kiro\\tools\\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: [] } } }参数对照表字段作用注意点command启动命令用python确保在 PATH 里args脚本路径必须绝对路径Windows 用双反斜杠env连接配置与脚本里的环境变量名一致disabled是否禁用设为 false 才会加载autoApprove自动批准的工具留空表示每次调用需确认注意args里的路径一定要用绝对路径。Kiro 的工作目录不一定是项目根目录用相对路径会直接报No such file or directory。4. 验证请求连通测试与消息收发配置写完后重启 Kiro让它重新加载 MCP Server。然后在对话里依次验证。第一步测试连接。输入“测试下 RabbitMQ 连接”Kiro 会调用test_connection()正常返回连接成功hostyour-host.com, port5672, vhost/第二步查队列状态。输入“查看 my-queue 队列的消息数”调用check_queue(my-queue){queue: my-queue, message_count: 3, consumer_count: 1}message_count是堆积数consumer_count是在线消费者数联调时一眼就能看出消费端有没有挂。第三步发消息。输入“发送一条测试消息到 my-exchange”调用publish_message(my-exchange, my-routing-key, {key: value})返回“消息发送成功”。然后回到第二步再查一次队列message_count应该加一。第四步读消息。输入“查看队列中的第一条消息”调用get_message(my-queue, auto_ackFalse)返回消息体 JSON。因为没删再查队列消息数不变说明“窥视”生效。如果这四步都通了说明 MCP Server 连通和消息收发都正常一次配置就跑通了。5. 本篇常见错排查问题一脚本路径找不到python: cant open file .kiro/tools/rabbitmq_mcp_server.py: No such file or directory原因是用了相对路径。Kiro 的工作目录不在项目根目录下改成绝对路径即可Windows 示例D:\\Project\\my-app\\.kiro\\tools\\rabbitmq_mcp_server.py。问题二PyPI 包依赖冲突ImportError: cannot import name BearerAuthProvider from fastmcp.server.auth这是开源 PyPI 包和fastmcp版本不兼容导致的。解决办法就是放弃第三方包用本文的自建本地脚本方案只依赖pika和mcp没有版本冲突。问题三国内网络拉取依赖超时Failed to fetch: https://pypi.org/simple/xxx - operation timed out可以换国内镜像源安装或者直接用本地脚本方案因为脚本本身不需要额外下载。问题四消费者反序列化失败Failed to invoke target method with argument type [class [B]这是 Java Spring 项目用SimpleMessageConverterJava 序列化导致的Python 发的 JSON 它反序列化不了。处理方式Java 序列化的消息通过 Java 单元测试发送Python MCP 只用来查看队列状态和消费已有消息不要用它发 Java 格式的消息。问题五连接被拒绝如果test_connection返回连接失败先确认 RabbitMQ 服务在跑、端口 5672 可达、用户名密码正确、vhost 存在。guest用户默认只能从 localhost 登录远程连接需要另建用户。6. 后续怎么用把 MCP 接进你的联调流程配置跑通之后这套东西的价值在于闭环。你可以让 Kiro 先通过 MySQL MCP 准备测试数据再用 RabbitMQ MCP 发消息触发业务逻辑最后查队列确认消费情况回到 MySQL 验证数据变化。整个过程在对话里完成不用来回切工具。如果你还想验证模型侧的调用是否正常可以去模型对话页https://taotoken.net/models试一下长期做编码和 Agent 任务的话Coding Planhttps://taotoken.net/coding-plan更合适接入文档在https://taotoken.net/docAPI Key 管理在https://taotoken.net/api-keys。把这几个地址存好下次换项目直接复用这套骨架就行。