Activemq Artemis MCP Server

AMQ Jolokia MCP服务器-源代码自述

概述

这是一个模型上下文协议（MCP）服务器，通过Jolokia API为Claude提供对Red Hat AMQ 7.12+的访问。它使用FastMCP简化服务器实现，使用aiohttp进行异步HTTP通信。

经过测试的AMQ版本

红帽AMQ 7.12.1

项目信息

• 姓名： AMQ Jolokia MCP服务器
• 类型： MCP服务器
• 框架： FastMCP
• 协议： 模型上下文协议（MCP）
• python 3.8+
• 紫外线： 0.8.4

建筑

文件结构

amq-jolokia-mcp-server/
├── amq-jolokia-server.py          # Main source file
├── requirements.txt       # Python dependencies
└── README.md             # This file

源代码组件

1.进口

import os                           # Environment variable access
import json                         # JSON serialization
import aiohttp                      # Async HTTP client
from typing import Optional         # Type hints
from mcp.server.fastmcp import FastMCP  # FastMCP framework

2.初始化

mcp = FastMCP("amq-jolokia-server")

创建名为的MCP服务器实例 amq-jolokia-server.

3.配置（环境变量）

AMQ_HOST = os.getenv("AMQ_HOST", "localhost")              # AMQ server hostname
AMQ_PORT = os.getenv("AMQ_PORT", "8161")                   # AMQ Jolokia port
AMQ_BROKER_NAME = os.getenv("AMQ_BROKER_NAME", "amq-broker-primary")  # Broker name
AMQ_ORIGIN = os.getenv("AMQ_ORIGIN", "mydomain.com")       # Origin header for CORS

BASE_URL = f"http://{AMQ_HOST}:{AMQ_PORT}/console/jolokia" # Jolokia API base URL

authenticated_credentials = {}  # Session storage for credentials

核心功能

call_jolokia_api()

签字：

async def call_jolokia_api(
    endpoint: str,
    method: str = "read",
    username: Optional[str] = None,
    password: Optional[str] = None,
    **params
) -> dict:

目的： 所有Jolokia API请求的通用处理程序

参数：

• endpoint （str）：Jolokia MBean对象名称

- 例子： org.apache.activemq.artemis:broker="amq-broker-primary"

• method （str）：Jolokia操作类型-“读取”、“写入”、“执行”或“搜索”
• username （可选\[str\]）：用于身份验证的AMQ代理用户名
• password （可选\[str\]）：用于身份验证的AMQ代理密码
• **params：其他URL参数（例如，属性=“版本”，操作=“浏览（）”）

逻辑流程：

1. URL构造：

   url = f"{BASE_URL}/{method}/{endpoint}"

- 将基本URL与方法和端点组合在一起 - 附加附加参数（如果提供） - 例子： http://localhost:8161/console/jolokia/read/org.apache.activemq.artemis:broker="..."/Version

1. 身份验证：

- 检查是否同时提供了用户名和密码 - 如果缺少凭据，则返回错误

1. HTTP请求：

   auth = aiohttp.BasicAuth(username, password)
   headers = {"Origin": AMQ_ORIGIN}
   async with aiohttp.ClientSession() as session:
       async with session.get(url, auth=auth, headers=headers) as response:

- 使用用户名/密码进行基本身份验证 - 为CORS合规性设置Origin标头 - 向Jolokia API发出异步GET请求

1. 响应处理：

- 以文本形式读取响应（处理Jolokia的文本/纯MIME类型） - 解析JSON响应 - 成功时返回解析的JSON（状态200） - 失败时返回错误字典

1. 错误处理：

- HTTP错误：捕获非200状态码 - JSON解析错误：处理无效的JSON响应 - 网络错误：捕获aiohttp的所有异常

退货：

{
    "error": "string",              # Error message if error occurred
    "message": "string",             # Additional error details
    "value": any,                    # Response data/payload
    "status": 200,                   # HTTP status code
    "timestamp": 1764119847,         # Server timestamp
    "request": {...}                 # Request metadata
}

MCP工具功能

工具定义使用 @mcp.tool() FastMCP的装饰师。

login(username: str, password: str) -> str

目的： 使用AMQ代理凭据对用户进行身份验证

实施：

1. 通过调用测试凭据 get_version API
2. 如果成功，则将凭据存储在 authenticated_credentials 字典
3. 返回成功/失败消息

用途：

login("admin", "admin-password")

退货：

"Successfully authenticated as user: admin"

logout() -> str

目的： 清除已验证的会话

实施：

1. 检查中是否存在凭据 authenticated_credentials
2. 清空字典
3. 返回注销确认

用途：

logout()

退货：

"Successfully logged out user: admin"

get_version() -> str

目的： 检索Red Hat AMQ代理版本

先决条件： 用户必须经过身份验证（首先呼叫登录）

实施：

1. 从会话中检索存储的用户名和密码
2. 验证身份验证状态
3. 为版本属性构建终结点： org.apache.activemq.artemis:broker="..."
4. 呼叫 call_jolokia_api() 方法=“读取”
5. 提取并返回版本字符串

用途：

get_version()

退货：

"AMQ Broker Version: 2.33.0.redhat-00013"

browse_queue(queue_name: str, routing_type: str = "anycast") -> str

目的： 浏览指定队列中的邮件

先决条件： 用户必须经过身份验证

参数：

• queue_name （str，必填）：要浏览的队列名称（例如，“HelloQueue”）
• routing_type （str，可选）：队列路由类型-“任意播”或“多播”（默认值：“anycast”）

实施：

1. 从会话中检索存储的凭据
2. 验证身份验证
3. 为队列构建复杂的MBean端点：

   org.apache.activemq.artemis:broker="...",component=addresses,address="...",
   subcomponent=queues,routing-type="...",queue="..."

1. 呼叫 call_jolokia_api() 方法=“exec”，操作=“browse（）”
2. 处理响应并提取消息数组
3. 返回带元数据的格式化JSON

用途：

browse_queue("HelloQueue", "anycast")

退货：

{
  "queue": "HelloQueue",
  "routing_type": "anycast",
  "message_count": 10,
  "messages": [
    {
      "messageID": "243788",
      "text": "hello3",
      "priority": 4,
      "timestamp": 1752717147852,
      "redelivered": false,
      "durable": true,
      "protocol": "CORE",
      "persistentSize": 234
    },
    ...
  ]
}

主入口点

if __name__ == "__main__":
    mcp.run()

• 启动FastMCP服务器
• 默认情况下使用stdio传输（与Claude和其他MCP客户端兼容）

依赖项

需求.txt

mcp>=0.1.0
aiohttp>=3.8.0

安装：

git clone https://github.com/ufoalan/ActiveMQ-Artemis-MCP-Server.git
cd ActiveMQ-Artemis-MCP-Server
uv venv
source .venv/bin/activate
uv pip install -r requirements.txt

依赖关系详细信息

使用流程

1.启动服务器

export AMQ_HOST=localhost
export AMQ_PORT=8161
export AMQ_BROKER_NAME=amq-broker-primary
export AMQ_ORIGIN=mydomain.com

uv run amq-jolokia-server.py

2.配置克劳德桌面

2a。编辑claude_desktop_config.json并在下面添加

{ “mcpServers”：{ “amq jolokia服务器”：{ “command”：“uv”， “args”：\[ “--目录”， “\”， “运行”， “amq jolokia服务器.py” \] } } }

2b。启动克劳德桌面

第一步：启动Claude桌面 步骤2：点击UI底部的“搜索和工具” 步骤3：检查菜单中是否出现“amq jolokia server”，应启用

3.在克劳德（或MCP客户端）

第一步：身份验证

login("admin", "admin-password")

步骤2：检查版本

get_version()

步骤3：浏览队列

browse_queue("HelloQueue")
browse_queue("OrderQueue", "anycast")

步骤4：注销

logout()

会话管理

凭据存储

凭据存储在模块级字典中：

authenticated_credentials = {
    "username": "admin",
    "password": "admin-password"
}

关键特性

• 在内存中： 服务器运行时仅存储在内存中的凭据
• 每个流程： 每个服务器进程都维护自己的凭据
• 基于会话： 凭据在以下时间清除 logout() 被称为
• 非持久性： 服务器重新启动时凭据丢失

安全说明

• 凭据仅用于对Jolokia的HTTP基本身份验证
• 没有将凭据记录或存储到磁盘
• 所有通信都使用HTTP（在生产环境中升级到HTTPS）
• 使用环境变量（非硬编码）进行配置

错误处理

错误响应格式

所有工具都以描述问题的字符串形式返回错误：

"Error: Authentication required - Please login first using the login tool"
"Error: HTTP 401 - Unauthorized"
"Error: Failed to browse queue - Queue not found"

常见错误案例

1. 未认证

- 触发器：登录前调用工具 - 响应：“错误：未通过身份验证。请先登录”

1. 无效凭证

- 触发器：登录时用户名/密码错误 - 响应：“身份验证失败：HTTP 401-…”

1. 未找到队列

- 触发器：队列名不存在的browse_queue - 响应：“浏览队列失败”

1. 连接错误

- 触发器：AMQ服务器无法访问 - 响应：“错误：\[连接错误详细信息\]”

配置示例

开发环境

export AMQ_HOST=localhost
export AMQ_PORT=8161
export AMQ_BROKER_NAME=amq-broker-primary
export AMQ_ORIGIN=localhost
uv run amq-jolokia-server.py

远程生产环境

export AMQ_HOST=amq-prod.example.com
export AMQ_PORT=8161
export AMQ_BROKER_NAME=amq-broker-primary
export AMQ_ORIGIN=example.com
uv run amq-jolokia-server.py

代码质量

• 类型提示： 用于函数参数和返回类型
• 文档字符串： 所有功能和工具的全面文档字符串
• 错误处理： 尝试使用catch块进行网络操作
• 异步/等待： I/O操作的正确异步实现
• 关注点分离： 具有特定工具实现的通用API处理程序

性能注意事项

• 异步I/O： 使用aiohttp的非阻塞HTTP请求
• 连接管理： aiohttp处理连接池
• 单次会话： 存储在内存中的凭据（适用于单个用户）
• 最低开销： FastMCP提供轻量级MCP服务器

测试

测试证书

# Verify connection
export AMQ_HOST=localhost
export AMQ_PORT=8161
python amq-jolokia-server.py

# In Claude/MCP client:
login("admin", "admin-password")
get_version()

调试提示

1. 添加打印语句以查看API调用
2. 检查 BASE_URL 正确： http://localhost:8161/console/jolokia
3. 使用手动curl验证凭据：

   curl -u admin:admin http://localhost:8161/console/jolokia/read/org.apache.activemq.artemis:broker=%22amq-broker-primary%22/Version

1. 确保 AMQ_ORIGIN 标头在AMQ服务器上被列入白名单

局限性

• 单凭据会话（一次一个用户）
• 存储在内存中的凭据（未持久化）
• 不支持SSL/TLS（使用HTTPS的反向代理）
• 无邮件过滤或分页
• 只读操作（仅浏览，不发送/删除）
• 登录时阻止凭据验证

未来的增强功能

• 多个并发会话
• 持久凭证存储
• SSL/TLS证书支持
• 消息发送/删除操作
• 队列统计和监控
• 邮件过滤和分页
• 速率限制
• 更好的错误恢复