feat(rabbitmq): add RabbitMQ integration (#6700)

* feat(rabbitmq): add RabbitMQ integration

* fix(rabbitmq): strip auth on redirect, require https, and bound the retrieval response

* fix(rabbitmq): reserve message metadata in the retrieval response budget
This commit is contained in:
Waleed
2026-08-14 12:05:01 -07:00
committed by GitHub
parent 3051954396
commit 1d342722ad
42 changed files with 5758 additions and 4 deletions
+16
View File
@@ -8578,6 +8578,22 @@ export function HexIcon(props: SVGProps<SVGSVGElement>) {
)
}
export function RabbitmqIcon(props: SVGProps<SVGSVGElement>) {
return (
<svg
{...props}
viewBox='-7.5 0 271 271'
preserveAspectRatio='xMidYMid'
xmlns='http://www.w3.org/2000/svg'
>
<path
d='M245.44 108.308h-85.09a7.738 7.738 0 0 1-7.735-7.734v-88.68C152.615 5.327 147.29 0 140.726 0h-30.375c-6.568 0-11.89 5.327-11.89 11.894v88.143c0 4.573-3.697 8.29-8.27 8.31l-27.885.133c-4.612.025-8.359-3.717-8.35-8.325l.173-88.241C54.144 5.337 48.817 0 42.24 0H11.89C5.321 0 0 5.327 0 11.894V260.21c0 5.834 4.726 10.56 10.555 10.56H245.44c5.834 0 10.56-4.726 10.56-10.56V118.868c0-5.834-4.726-10.56-10.56-10.56zm-39.902 93.233c0 7.645-6.198 13.844-13.843 13.844H167.69c-7.646 0-13.844-6.199-13.844-13.844v-24.005c0-7.646 6.198-13.844 13.844-13.844h24.005c7.645 0 13.843 6.198 13.843 13.844v24.005z'
fill='#F60'
/>
</svg>
)
}
export function RailwayIcon(props: SVGProps<SVGSVGElement>) {
return (
<svg {...props} xmlns='http://www.w3.org/2000/svg' fill='currentColor' viewBox='0 0 24 24'>
+2
View File
@@ -182,6 +182,7 @@ import {
QdrantIcon,
QuartrIcon,
QuiverIcon,
RabbitmqIcon,
RailwayIcon,
RB2BIcon,
RDSIcon,
@@ -469,6 +470,7 @@ export const blockTypeToIconMap: Record<string, IconComponent> = {
qdrant: QdrantIcon,
quartr: QuartrIcon,
quiver: QuiverIcon,
rabbitmq: RabbitmqIcon,
railway: RailwayIcon,
rb2b: RB2BIcon,
rds: RDSIcon,
@@ -194,6 +194,7 @@
"qdrant",
"quartr",
"quiver",
"rabbitmq",
"railway",
"rb2b",
"rds",
@@ -0,0 +1,564 @@
---
title: RabbitMQ
description: Publish and read messages and manage queues in RabbitMQ
---
import { BlockInfoCard } from "@/components/ui/block-info-card"
<BlockInfoCard
type="rabbitmq"
color="#FFFFFF"
/>
{/* MANUAL-CONTENT-START:intro */}
[RabbitMQ](https://www.rabbitmq.com/) is an open-source message broker that sits between the parts of a system that produce work and the parts that do it. A producer publishes a message to an **exchange**, the exchange matches the message's **routing key** against its **bindings**, and every matching **queue** holds the message until a consumer takes it. That indirection is the point: producers never need to know who consumes their messages, and a queue absorbs bursts that would otherwise overwhelm a downstream service.
**Why RabbitMQ?**
- **Durable buffering:** A queue holds work while consumers are slow, restarting, or offline, so a traffic spike becomes a backlog to work through rather than dropped requests.
- **Flexible routing:** Direct exchanges route on an exact key, topic exchanges on wildcard patterns like `orders.*`, fanout exchanges to every bound queue, and headers exchanges on message metadata.
- **Delivery guarantees:** Messages and queues can be marked durable so they survive a broker restart, and unacknowledged messages return to the queue when a consumer dies mid-work.
- **Failure handling built in:** Dead-letter exchanges, per-message TTLs, and queue length limits let you decide up front what happens to work that expires, overflows, or repeatedly fails.
- **Runs anywhere:** Self-hosted on your own infrastructure, or managed through providers such as CloudAMQP and Amazon MQ.
**Using RabbitMQ in Sim**
Sim talks to RabbitMQ over its **Management HTTP API** — the same interface behind the RabbitMQ management UI — using the management plugin's base URL plus a username and password. There is no AMQP connection to configure and no client library to install; if you can reach the management UI in a browser, Sim can reach your broker.
**Key benefits of using RabbitMQ in Sim:**
- **Publish from any workflow step:** Hand off enriched or classified data to an existing service by publishing to an exchange, without that service needing to know Sim exists.
- **Inspect queues without consuming them:** The default acknowledgement mode requeues what it reads, so an agent can examine a dead-letter backlog and leave the messages exactly where they were.
- **Triage failures with an agent:** Read a dead-letter queue, let an agent group messages by root cause, and route the summary to Slack, PagerDuty, or a table.
- **Monitor broker health on a schedule:** List queues and read the broker overview to catch a queue whose depth is climbing or that has lost all its consumers.
- **Declare topology as part of a workflow:** Create queues, set arguments such as quorum type or dead-lettering, and bind them to exchanges as an automated setup step.
**Before you start**
- The **management plugin must be enabled and reachable** from Sim. It listens on port `15672` by default and is separate from the AMQP port (`5672`). Self-hosted brokers enable it with `rabbitmq-plugins enable rabbitmq_management`; managed providers expose it as a management or console URL.
- The management URL **must use `https`** unless the broker is on a loopback host. Credentials travel on every request as HTTP basic auth, so plain `http` to a remote broker would put them on the wire in the clear — Sim rejects it rather than sending them.
- The user you authenticate as needs the **`management` tag** at minimum, plus read and write permissions on the virtual host you target. Administrative operations require broader permissions.
- Publishing and reading messages over the HTTP API is **convenient but not a high-throughput transport** — RabbitMQ opens a new connection per request. It is well suited to workflow-rate traffic, inspection, and operational automation; a service consuming thousands of messages per second should use an AMQP client instead.
- Queue statistics such as message and consumer counts are **collected on an interval**, so a queue declared moments ago may report them as empty until the broker's next sample.
- Reading messages is **bounded per call** so one retrieval cannot exceed Sim's response limit. A batch is capped at 50 messages, payloads are truncated (each message reports whether it was), and a large batch shortens payloads further. AMQP properties and headers are returned in full — the broker offers no way to truncate them — so retrieving several messages carrying very large headers may still hit the limit; lower the count if that happens.
{/* MANUAL-CONTENT-END */}
## Usage Instructions
Connect agents to a RabbitMQ broker through its Management HTTP API. Publish messages to exchanges, read messages off queues, declare queues, exchanges, bindings, and policies, and inspect broker health, queue depth, consumers, connections, and cluster nodes. Works with self-hosted brokers and managed offerings such as CloudAMQP as long as the management plugin is reachable.
## Actions
### RabbitMQ Publish Message
Publish a message to a RabbitMQ exchange with a routing key. Reports whether the message was routed to at least one queue.
#### Input
| Parameter | Type | Required | Description |
| --------- | ---- | -------- | ----------- |
| `exchange` | string | No | Exchange to publish to. Leave empty to publish to the default exchange, which routes by queue name. Empty is a valid value, so this is not required. |
| `routingKey` | string | Yes | Routing key. When publishing to the default exchange this is the target queue name. |
| `payload` | string | Yes | Message body to publish |
| `payloadEncoding` | string | No | How the payload is encoded: string \(default\) or base64 |
| `properties` | string | No | AMQP basic properties as a JSON object, e.g. \{"delivery_mode":2,"content_type":"application/json"\} |
| `headers` | string | No | Message headers as a JSON object, e.g. \{"source":"sim"\} |
#### Output
| Parameter | Type | Description |
| --------- | ---- | ----------- |
| `routed` | boolean | Whether the message was routed to at least one queue. False means no binding matched and the message was dropped. |
| `exchange` | string | Exchange the message was published to |
| `routingKey` | string | Routing key the message was published with |
### RabbitMQ Get Messages
Retrieve messages from a RabbitMQ queue. Defaults to requeueing the messages so they stay available to real consumers.
#### Input
| Parameter | Type | Required | Description |
| --------- | ---- | -------- | ----------- |
| `queue` | string | Yes | Queue to read messages from |
| `count` | number | No | Maximum number of messages to retrieve, from 1 to $\{MAX_MESSAGE_COUNT\}. Defaults to 1 |
| `ackmode` | string | No | How retrieved messages are handled: ack_requeue_true \(default, leaves messages in the queue\), ack_requeue_false \(removes them\), reject_requeue_true, or reject_requeue_false |
| `encoding` | string | No | auto \(default\) returns readable text where possible, base64 always returns base64 |
| `truncate` | number | No | Truncate payloads longer than this many bytes. Defaults to $\{DEFAULT_TRUNCATE_BYTES\}, capped at $\{MAX_TRUNCATE_BYTES\}, and lowered further at high counts so the whole batch stays inside the response limit. Each message reports whether it was truncated |
#### Output
| Parameter | Type | Description |
| --------- | ---- | ----------- |
| `queueName` | string | Queue the messages were read from |
| `count` | number | Number of messages retrieved |
| `messages` | array | Retrieved messages, empty when the queue holds nothing |
### RabbitMQ List Queues
List queues in a RabbitMQ virtual host with their depth, consumer count, and configuration.
#### Input
| Parameter | Type | Required | Description |
| --------- | ---- | -------- | ----------- |
| `page` | number | No | Page of results to return, starting at 1 |
| `pageSize` | number | No | Queues per page, from 1 to $\{RABBITMQ_MAX_PAGE_SIZE\}. Defaults to $\{DEFAULT_PAGE_SIZE\} |
| `name` | string | No | Filter queues whose name contains this value |
| `useRegex` | boolean | No | Treat the name filter as a regular expression |
#### Output
| Parameter | Type | Description |
| --------- | ---- | ----------- |
| `queues` | array | Queues in the virtual host |
| `count` | number | Number of queues returned on this page |
| `totalCount` | number | Total queues in the virtual host before filtering |
| `page` | number | Page number returned |
| `pageCount` | number | Total number of pages |
### RabbitMQ Get Queue
Read a single RabbitMQ queue, including its depth, consumer count, and declaration settings.
#### Input
| Parameter | Type | Required | Description |
| --------- | ---- | -------- | ----------- |
| `queue` | string | Yes | Queue name to read |
#### Output
| Parameter | Type | Description |
| --------- | ---- | ----------- |
| `queue` | object | The requested queue |
### RabbitMQ Create Queue
Declare a RabbitMQ queue. Declaring a queue that already exists with the same settings succeeds without changing it.
#### Input
| Parameter | Type | Required | Description |
| --------- | ---- | -------- | ----------- |
| `queue` | string | Yes | Name of the queue to declare |
| `durable` | boolean | No | Whether the queue survives a broker restart. Defaults to true |
| `autoDelete` | boolean | No | Delete the queue when its last consumer disconnects. Defaults to false |
| `arguments` | string | No | Queue arguments as a JSON object, e.g. \{"x-queue-type":"quorum","x-message-ttl":60000\} |
#### Output
| Parameter | Type | Description |
| --------- | ---- | ----------- |
| `queueName` | string | Name of the declared queue |
| `vhost` | string | Virtual host the queue was declared in |
| `created` | boolean | Whether the declaration succeeded |
### RabbitMQ Delete Queue
Delete a RabbitMQ queue and every message still in it. Can be guarded so the delete only happens when the queue is unused or empty.
#### Input
| Parameter | Type | Required | Description |
| --------- | ---- | -------- | ----------- |
| `queue` | string | Yes | Name of the queue to delete |
| `ifUnused` | boolean | No | Only delete the queue when it has no consumers |
| `ifEmpty` | boolean | No | Only delete the queue when it holds no messages |
#### Output
| Parameter | Type | Description |
| --------- | ---- | ----------- |
| `queueName` | string | Name of the deleted queue |
| `vhost` | string | Virtual host the queue was deleted from |
| `deleted` | boolean | Whether the queue was deleted |
### RabbitMQ Purge Queue
Discard every ready message in a RabbitMQ queue while leaving the queue itself in place.
#### Input
| Parameter | Type | Required | Description |
| --------- | ---- | -------- | ----------- |
| `queue` | string | Yes | Name of the queue to purge |
#### Output
| Parameter | Type | Description |
| --------- | ---- | ----------- |
| `queueName` | string | Name of the purged queue |
| `vhost` | string | Virtual host the queue belongs to |
| `purged` | boolean | Whether the queue was purged |
### RabbitMQ List Exchanges
List exchanges in a RabbitMQ virtual host with their type and declaration settings.
#### Input
| Parameter | Type | Required | Description |
| --------- | ---- | -------- | ----------- |
| `page` | number | No | Page of results to return, starting at 1 |
| `pageSize` | number | No | Exchanges per page, from 1 to $\{RABBITMQ_MAX_PAGE_SIZE\}. Defaults to $\{DEFAULT_PAGE_SIZE\} |
| `name` | string | No | Filter exchanges whose name contains this value |
| `useRegex` | boolean | No | Treat the name filter as a regular expression |
#### Output
| Parameter | Type | Description |
| --------- | ---- | ----------- |
| `exchanges` | array | Exchanges in the virtual host |
| `count` | number | Number of exchanges returned on this page |
| `totalCount` | number | Total exchanges in the virtual host before filtering |
| `page` | number | Page number returned |
| `pageCount` | number | Total number of pages |
### RabbitMQ Get Exchange
Read a single RabbitMQ exchange and the settings it was declared with.
#### Input
| Parameter | Type | Required | Description |
| --------- | ---- | -------- | ----------- |
| `exchange` | string | No | Exchange name to read. Leave empty for the default exchange, which is a valid value, so this is not required |
#### Output
| Parameter | Type | Description |
| --------- | ---- | ----------- |
| `exchange` | object | The requested exchange |
### RabbitMQ Create Exchange
Declare a RabbitMQ exchange. Declaring an exchange that already exists with the same settings succeeds without changing it.
#### Input
| Parameter | Type | Required | Description |
| --------- | ---- | -------- | ----------- |
| `exchange` | string | Yes | Name of the exchange to declare |
| `exchangeType` | string | No | Routing behaviour: direct \(exact routing key, default\), topic \(wildcard patterns\), fanout \(every bound queue\), or headers \(match on binding arguments\) |
| `durable` | boolean | No | Whether the exchange survives a broker restart. Defaults to true |
| `autoDelete` | boolean | No | Delete the exchange once its last binding is removed. Defaults to false |
| `internal` | boolean | No | Internal exchanges cannot be published to directly, only bound from another exchange. Defaults to false |
| `arguments` | string | No | Exchange arguments as a JSON object, e.g. \{"alternate-exchange":"unrouted"\} to capture messages that match no binding |
#### Output
| Parameter | Type | Description |
| --------- | ---- | ----------- |
| `exchangeName` | string | Name of the declared exchange |
| `vhost` | string | Virtual host the exchange was declared in |
| `created` | boolean | Whether the declaration succeeded |
### RabbitMQ Delete Exchange
Delete a RabbitMQ exchange and every binding attached to it. Publishers targeting it will fail afterwards.
#### Input
| Parameter | Type | Required | Description |
| --------- | ---- | -------- | ----------- |
| `exchange` | string | Yes | Name of the exchange to delete |
| `ifUnused` | boolean | No | Only delete the exchange when nothing is bound to it |
#### Output
| Parameter | Type | Description |
| --------- | ---- | ----------- |
| `exchangeName` | string | Name of the deleted exchange |
| `vhost` | string | Virtual host the exchange was deleted from |
| `deleted` | boolean | Whether the exchange was deleted |
### RabbitMQ List Bindings
List the bindings that route messages into a RabbitMQ queue, including the implicit default-exchange binding.
#### Input
| Parameter | Type | Required | Description |
| --------- | ---- | -------- | ----------- |
| `queue` | string | Yes | Queue whose bindings should be listed |
#### Output
| Parameter | Type | Description |
| --------- | ---- | ----------- |
| `queueName` | string | Queue the bindings route into |
| `bindings` | array | Bindings targeting the queue. The entry with an empty source is the implicit default-exchange binding |
| `count` | number | Number of bindings returned |
### RabbitMQ List Exchange Bindings
List everything an exchange routes to, so you can see which routing keys reach which queues.
#### Input
| Parameter | Type | Required | Description |
| --------- | ---- | -------- | ----------- |
| `exchange` | string | No | Exchange whose outgoing bindings should be listed. Leave empty for the default exchange, which is a valid value, so this is not required |
#### Output
| Parameter | Type | Description |
| --------- | ---- | ----------- |
| `exchangeName` | string | Exchange the bindings originate from |
| `bindings` | array | Bindings routing out of the exchange. An empty list means nothing it publishes can be delivered |
| `count` | number | Number of bindings returned |
### RabbitMQ Create Binding
Bind a queue or another exchange to a RabbitMQ exchange so messages matching a routing key are routed to it.
#### Input
| Parameter | Type | Required | Description |
| --------- | ---- | -------- | ----------- |
| `exchange` | string | Yes | Source exchange to bind from |
| `queue` | string | Yes | Destination queue, or destination exchange when binding exchange to exchange |
| `destinationType` | string | No | Whether the destination is a queue \(default\) or an exchange. Exchange-to-exchange bindings chain routing between exchanges |
| `routingKey` | string | No | Routing key the binding matches. Topic exchanges accept wildcards such as orders.* |
| `arguments` | string | No | Binding arguments as a JSON object. Headers exchanges match on these, e.g. \{"x-match":"all","type":"invoice"\} |
#### Output
| Parameter | Type | Description |
| --------- | ---- | ----------- |
| `exchange` | string | Source exchange the binding reads from |
| `queueName` | string | Destination queue the binding routes into |
| `routingKey` | string | Routing key the binding matches |
| `propertiesKey` | string | Broker identifier addressing the new binding |
| `created` | boolean | Whether the binding was created |
### RabbitMQ Delete Binding
Remove a binding so an exchange stops routing its matching messages to that destination.
#### Input
| Parameter | Type | Required | Description |
| --------- | ---- | -------- | ----------- |
| `exchange` | string | Yes | Source exchange the binding reads from |
| `destination` | string | Yes | Destination queue or exchange the binding routes to |
| `destinationType` | string | No | Whether the destination is a queue \(default\) or an exchange |
| `propertiesKey` | string | Yes | Broker identifier for the binding, taken from List Bindings or Create Binding. It is the routing key for a simple binding, ~ for an empty routing key, and a hashed value when the binding has arguments |
#### Output
| Parameter | Type | Description |
| --------- | ---- | ----------- |
| `exchange` | string | Source exchange the binding read from |
| `destination` | string | Destination the binding routed to |
| `propertiesKey` | string | Broker identifier of the deleted binding |
| `deleted` | boolean | Whether the binding was deleted |
### RabbitMQ Get Overview
Read broker-wide RabbitMQ status: version, cluster name, object totals, queue depth totals, and message rates.
#### Input
| Parameter | Type | Required | Description |
| --------- | ---- | -------- | ----------- |
#### Output
| Parameter | Type | Description |
| --------- | ---- | ----------- |
| `rabbitmqVersion` | string | RabbitMQ version running on the node |
| `productName` | string | Broker product name |
| `productVersion` | string | Broker product version |
| `erlangVersion` | string | Erlang runtime version |
| `clusterName` | string | Name of the cluster |
| `node` | string | Node that served the request |
| `objectTotals` | object | Counts of brokers objects |
| ↳ `connections` | number | Open connections |
| ↳ `channels` | number | Open channels |
| ↳ `exchanges` | number | Declared exchanges |
| ↳ `queues` | number | Declared queues |
| ↳ `consumers` | number | Registered consumers |
| `queueTotals` | object | Aggregate queue depth across the broker |
| ↳ `messages` | number | Total messages across all queues |
| ↳ `messages_ready` | number | Messages ready for delivery |
| ↳ `messages_unacknowledged` | number | Delivered but unacknowledged messages |
| `messageStats` | json | Broker-wide message counters and rates, e.g. publish and confirm totals |
### RabbitMQ Health Check
Run one of the broker health checks and report whether it passed. A failing check is a normal result, not a tool error.
#### Input
| Parameter | Type | Required | Description |
| --------- | ---- | -------- | ----------- |
| `check` | string | No | Which check to run: alarms \(cluster-wide resource alarms, default\), local-alarms, virtual-hosts, node-is-quorum-critical, port-listener, protocol-listener, or certificate-expiration |
| `port` | number | No | Port to verify a listener on. Required for the port-listener check |
| `protocol` | string | No | Protocol to verify a listener for, e.g. amqp, amqp/ssl, mqtt, stomp, or http. Required for the protocol-listener check |
| `within` | number | No | How far ahead to look for expiring certificates. Required for the certificate-expiration check |
| `unit` | string | No | Unit for the certificate-expiration window: days, weeks, months \(default\), or years |
#### Output
| Parameter | Type | Description |
| --------- | ---- | ----------- |
| `check` | string | The health check that was run |
| `healthy` | boolean | True when the check reported status ok |
| `status` | string | Raw status reported by the broker: ok or failed |
| `reason` | string | Explanation the broker gave, present on failures and on some passes |
| `details` | json | Full check body, including check-specific fields such as the ports or protocols found |
### RabbitMQ List Nodes
List the cluster nodes with memory, disk, file-descriptor, and alarm state. A fired alarm blocks publishers broker-wide.
#### Input
| Parameter | Type | Required | Description |
| --------- | ---- | -------- | ----------- |
#### Output
| Parameter | Type | Description |
| --------- | ---- | ----------- |
| `nodes` | array | Cluster nodes and their resource headroom |
| `count` | number | Number of nodes in the cluster |
### RabbitMQ List Virtual Hosts
List the virtual hosts on the broker with their message totals, so you can discover which scopes exist.
#### Input
| Parameter | Type | Required | Description |
| --------- | ---- | -------- | ----------- |
#### Output
| Parameter | Type | Description |
| --------- | ---- | ----------- |
| `vhosts` | array | Virtual hosts the authenticated user can see |
| `count` | number | Number of virtual hosts returned |
### RabbitMQ List Connections
List client connections to the broker with their user, state, and channel count. Connections are cluster-wide, not scoped to one virtual host.
#### Input
| Parameter | Type | Required | Description |
| --------- | ---- | -------- | ----------- |
| `page` | number | No | Page of results to return, starting at 1 |
| `pageSize` | number | No | Connections per page, from 1 to $\{RABBITMQ_MAX_PAGE_SIZE\}. Defaults to $\{DEFAULT_PAGE_SIZE\} |
| `name` | string | No | Filter connections whose name contains this value |
| `useRegex` | boolean | No | Treat the name filter as a regular expression |
#### Output
| Parameter | Type | Description |
| --------- | ---- | ----------- |
| `connections` | array | Open client connections |
| `count` | number | Number of connections returned on this page |
| `totalCount` | number | Total connections before filtering |
| `page` | number | Page number returned |
| `pageCount` | number | Total number of pages |
### RabbitMQ List Channels
List open channels with their prefetch limit and unacknowledged message count, which is where stalled consumers show up. Channels are cluster-wide, not scoped to one virtual host.
#### Input
| Parameter | Type | Required | Description |
| --------- | ---- | -------- | ----------- |
| `page` | number | No | Page of results to return, starting at 1 |
| `pageSize` | number | No | Channels per page, from 1 to $\{RABBITMQ_MAX_PAGE_SIZE\}. Defaults to $\{DEFAULT_PAGE_SIZE\} |
| `name` | string | No | Filter channels whose name contains this value |
| `useRegex` | boolean | No | Treat the name filter as a regular expression |
#### Output
| Parameter | Type | Description |
| --------- | ---- | ----------- |
| `channels` | array | Open channels |
| `count` | number | Number of channels returned on this page |
| `totalCount` | number | Total channels before filtering |
| `page` | number | Page number returned |
| `pageCount` | number | Total number of pages |
### RabbitMQ List Consumers
List the consumers subscribed in a virtual host. An empty result for a queue with a backlog means nothing is processing it.
#### Input
| Parameter | Type | Required | Description |
| --------- | ---- | -------- | ----------- |
#### Output
| Parameter | Type | Description |
| --------- | ---- | ----------- |
| `consumers` | array | Consumers currently subscribed in the virtual host |
| `count` | number | Number of consumers returned |
### RabbitMQ List Policies
List the policies in a virtual host. Policies are how dead-lettering, TTLs, and length limits get applied to matching queues and exchanges.
#### Input
| Parameter | Type | Required | Description |
| --------- | ---- | -------- | ----------- |
#### Output
| Parameter | Type | Description |
| --------- | ---- | ----------- |
| `policies` | array | Policies defined in the virtual host |
| `count` | number | Number of policies returned |
### RabbitMQ Create Policy
Create or replace a RabbitMQ policy, applying settings such as dead-lettering, TTLs, or length limits to every queue or exchange whose name matches a pattern.
#### Input
| Parameter | Type | Required | Description |
| --------- | ---- | -------- | ----------- |
| `policyName` | string | Yes | Name of the policy. Reusing an existing name replaces that policy |
| `pattern` | string | Yes | Regular expression matched against queue or exchange names, e.g. ^orders\\. to match every name starting with orders. |
| `definition` | string | Yes | Settings to apply, as a JSON object, e.g. \{"dead-letter-exchange":"dlx","message-ttl":86400000,"max-length":10000\} |
| `priority` | number | No | Priority, defaulting to 0. When several policies match a resource only the highest-priority one applies — they do not merge |
| `applyTo` | string | No | What the policy applies to: queues \(default\), classic_queues, quorum_queues, streams, exchanges, or all |
#### Output
| Parameter | Type | Description |
| --------- | ---- | ----------- |
| `policyName` | string | Name of the created policy |
| `vhost` | string | Virtual host the policy applies in |
| `created` | boolean | Whether the policy was created or replaced |
### RabbitMQ Delete Policy
Delete a RabbitMQ policy. Every queue and exchange it matched immediately loses the settings it applied.
#### Input
| Parameter | Type | Required | Description |
| --------- | ---- | -------- | ----------- |
| `policyName` | string | Yes | Name of the policy to delete |
#### Output
| Parameter | Type | Description |
| --------- | ---- | ----------- |
| `policyName` | string | Name of the deleted policy |
| `vhost` | string | Virtual host the policy applied in |
| `deleted` | boolean | Whether the policy was deleted |
File diff suppressed because it is too large Load Diff
+3
View File
@@ -251,6 +251,7 @@ import { PulseBlock, PulseBlockMeta, PulseV2Block } from '@/blocks/blocks/pulse'
import { QdrantBlock, QdrantBlockMeta } from '@/blocks/blocks/qdrant'
import { QuartrBlock, QuartrBlockMeta } from '@/blocks/blocks/quartr'
import { QuiverBlock, QuiverBlockMeta } from '@/blocks/blocks/quiver'
import { RabbitmqBlock, RabbitmqBlockMeta } from '@/blocks/blocks/rabbitmq'
import { RailwayBlock, RailwayBlockMeta } from '@/blocks/blocks/railway'
import { RB2BBlock, RB2BBlockMeta } from '@/blocks/blocks/rb2b'
import { RDSBlock, RDSBlockMeta } from '@/blocks/blocks/rds'
@@ -575,6 +576,7 @@ export const BLOCK_REGISTRY: Record<string, BlockConfig> = {
qdrant: QdrantBlock,
quartr: QuartrBlock,
quiver: QuiverBlock,
rabbitmq: RabbitmqBlock,
railway: RailwayBlock,
rb2b: RB2BBlock,
rds: RDSBlock,
@@ -877,6 +879,7 @@ export const BLOCK_META_REGISTRY: Record<string, BlockMeta> = {
qdrant: QdrantBlockMeta,
quartr: QuartrBlockMeta,
quiver: QuiverBlockMeta,
rabbitmq: RabbitmqBlockMeta,
railway: RailwayBlockMeta,
rb2b: RB2BBlockMeta,
rds: RDSBlockMeta,
+16
View File
@@ -8578,6 +8578,22 @@ export function HexIcon(props: SVGProps<SVGSVGElement>) {
)
}
export function RabbitmqIcon(props: SVGProps<SVGSVGElement>) {
return (
<svg
{...props}
viewBox='-7.5 0 271 271'
preserveAspectRatio='xMidYMid'
xmlns='http://www.w3.org/2000/svg'
>
<path
d='M245.44 108.308h-85.09a7.738 7.738 0 0 1-7.735-7.734v-88.68C152.615 5.327 147.29 0 140.726 0h-30.375c-6.568 0-11.89 5.327-11.89 11.894v88.143c0 4.573-3.697 8.29-8.27 8.31l-27.885.133c-4.612.025-8.359-3.717-8.35-8.325l.173-88.241C54.144 5.337 48.817 0 42.24 0H11.89C5.321 0 0 5.327 0 11.894V260.21c0 5.834 4.726 10.56 10.555 10.56H245.44c5.834 0 10.56-4.726 10.56-10.56V118.868c0-5.834-4.726-10.56-10.56-10.56zm-39.902 93.233c0 7.645-6.198 13.844-13.843 13.844H167.69c-7.646 0-13.844-6.199-13.844-13.844v-24.005c0-7.646 6.198-13.844 13.844-13.844h24.005c7.645 0 13.843 6.198 13.843 13.844v24.005z'
fill='#F60'
/>
</svg>
)
}
export function RailwayIcon(props: SVGProps<SVGSVGElement>) {
return (
<svg {...props} xmlns='http://www.w3.org/2000/svg' fill='currentColor' viewBox='0 0 24 24'>
@@ -180,6 +180,7 @@ import {
QdrantIcon,
QuartrIcon,
QuiverIcon,
RabbitmqIcon,
RailwayIcon,
RB2BIcon,
RDSIcon,
@@ -451,6 +452,7 @@ export const blockTypeToIconMap: Record<string, IconComponent> = {
qdrant: QdrantIcon,
quartr: QuartrIcon,
quiver: QuiverIcon,
rabbitmq: RabbitmqIcon,
railway: RailwayIcon,
rb2b: RB2BIcon,
rds: RDSIcon,
+120 -1
View File
@@ -1,5 +1,5 @@
{
"updatedAt": "2026-08-13",
"updatedAt": "2026-08-14",
"integrations": [
{
"type": "onepassword",
@@ -15203,6 +15203,125 @@
"integrationType": "ai",
"tags": ["image-generation"]
},
{
"type": "rabbitmq",
"slug": "rabbitmq",
"name": "RabbitMQ",
"description": "Publish and read messages and manage queues in RabbitMQ",
"longDescription": "Connect agents to a RabbitMQ broker through its Management HTTP API. Publish messages to exchanges, read messages off queues, declare queues, exchanges, bindings, and policies, and inspect broker health, queue depth, consumers, connections, and cluster nodes. Works with self-hosted brokers and managed offerings such as CloudAMQP as long as the management plugin is reachable.",
"bgColor": "#FFFFFF",
"iconName": "RabbitmqIcon",
"docsUrl": "https://docs.sim.ai/integrations/rabbitmq",
"operations": [
{
"name": "Publish Message",
"description": "Publish a message to a RabbitMQ exchange with a routing key. Reports whether the message was routed to at least one queue."
},
{
"name": "Get Messages",
"description": "Retrieve messages from a RabbitMQ queue. Defaults to requeueing the messages so they stay available to real consumers."
},
{
"name": "List Queues",
"description": "List queues in a RabbitMQ virtual host with their depth, consumer count, and configuration."
},
{
"name": "Get Queue",
"description": "Read a single RabbitMQ queue, including its depth, consumer count, and declaration settings."
},
{
"name": "Create Queue",
"description": "Declare a RabbitMQ queue. Declaring a queue that already exists with the same settings succeeds without changing it."
},
{
"name": "Delete Queue",
"description": "Delete a RabbitMQ queue and every message still in it. Can be guarded so the delete only happens when the queue is unused or empty."
},
{
"name": "Purge Queue",
"description": "Discard every ready message in a RabbitMQ queue while leaving the queue itself in place."
},
{
"name": "List Exchanges",
"description": "List exchanges in a RabbitMQ virtual host with their type and declaration settings."
},
{
"name": "Get Exchange",
"description": "Read a single RabbitMQ exchange and the settings it was declared with."
},
{
"name": "Create Exchange",
"description": "Declare a RabbitMQ exchange. Declaring an exchange that already exists with the same settings succeeds without changing it."
},
{
"name": "Delete Exchange",
"description": "Delete a RabbitMQ exchange and every binding attached to it. Publishers targeting it will fail afterwards."
},
{
"name": "List Queue Bindings",
"description": "List the bindings that route messages into a RabbitMQ queue, including the implicit default-exchange binding."
},
{
"name": "List Exchange Bindings",
"description": "List everything an exchange routes to, so you can see which routing keys reach which queues."
},
{
"name": "Create Binding",
"description": "Bind a queue or another exchange to a RabbitMQ exchange so messages matching a routing key are routed to it."
},
{
"name": "Delete Binding",
"description": "Remove a binding so an exchange stops routing its matching messages to that destination."
},
{
"name": "Get Overview",
"description": "Read broker-wide RabbitMQ status: version, cluster name, object totals, queue depth totals, and message rates."
},
{
"name": "Health Check",
"description": "Run one of the broker health checks and report whether it passed. A failing check is a normal result, not a tool error."
},
{
"name": "List Nodes",
"description": "List the cluster nodes with memory, disk, file-descriptor, and alarm state. A fired alarm blocks publishers broker-wide."
},
{
"name": "List Virtual Hosts",
"description": "List the virtual hosts on the broker with their message totals, so you can discover which scopes exist."
},
{
"name": "List Connections",
"description": "List client connections to the broker with their user, state, and channel count. Connections are cluster-wide, not scoped to one virtual host."
},
{
"name": "List Channels",
"description": "List open channels with their prefetch limit and unacknowledged message count, which is where stalled consumers show up. Channels are cluster-wide, not scoped to one virtual host."
},
{
"name": "List Consumers",
"description": "List the consumers subscribed in a virtual host. An empty result for a queue with a backlog means nothing is processing it."
},
{
"name": "List Policies",
"description": "List the policies in a virtual host. Policies are how dead-lettering, TTLs, and length limits get applied to matching queues and exchanges."
},
{
"name": "Create Policy",
"description": "Create or replace a RabbitMQ policy, applying settings such as dead-lettering, TTLs, or length limits to every queue or exchange whose name matches a pattern."
},
{
"name": "Delete Policy",
"description": "Delete a RabbitMQ policy. Every queue and exchange it matched immediately loses the settings it applied."
}
],
"operationCount": 25,
"triggers": [],
"triggerCount": 0,
"authType": "api-key",
"category": "tools",
"integrationType": "devops",
"tags": ["messaging", "automation"]
},
{
"type": "railway",
"slug": "railway",
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
+136
View File
@@ -0,0 +1,136 @@
import type {
RabbitmqCreateBindingParams,
RabbitmqCreateBindingResponse,
} from '@/tools/rabbitmq/types'
import {
buildAuthHeaders,
buildManagementUrl,
extractErrorMessage,
parseJsonObjectParam,
RABBITMQ_CONNECTION_PARAMS,
resolveVhost,
} from '@/tools/rabbitmq/utils'
import type { ToolConfig } from '@/tools/types'
/**
* The broker returns the new binding as a `location` header shaped `destination/properties_key`,
* with the properties key percent-encoded. The properties key never contains a literal slash,
* so the final segment is the key.
*/
function extractPropertiesKey(location: string | null): string | null {
if (!location) return null
const segment = location.slice(location.lastIndexOf('/') + 1)
if (!segment) return null
try {
return decodeURIComponent(segment)
} catch {
return segment
}
}
export const rabbitmqCreateBindingTool: ToolConfig<
RabbitmqCreateBindingParams,
RabbitmqCreateBindingResponse
> = {
id: 'rabbitmq_create_binding',
name: 'RabbitMQ Create Binding',
description:
'Bind a queue or another exchange to a RabbitMQ exchange so messages matching a routing key are routed to it.',
version: '1.0.0',
params: {
...RABBITMQ_CONNECTION_PARAMS,
exchange: {
type: 'string',
required: true,
visibility: 'user-or-llm',
description: 'Source exchange to bind from',
},
queue: {
type: 'string',
required: true,
visibility: 'user-or-llm',
description: 'Destination queue, or destination exchange when binding exchange to exchange',
},
destinationType: {
type: 'string',
required: false,
visibility: 'user-or-llm',
description:
'Whether the destination is a queue (default) or an exchange. Exchange-to-exchange bindings chain routing between exchanges',
},
routingKey: {
type: 'string',
required: false,
visibility: 'user-or-llm',
description:
'Routing key the binding matches. Topic exchanges accept wildcards such as orders.*',
},
arguments: {
type: 'string',
required: false,
visibility: 'user-or-llm',
description:
'Binding arguments as a JSON object. Headers exchanges match on these, e.g. {"x-match":"all","type":"invoice"}',
},
},
request: {
url: ({ host, vhost, destinationType, exchange, queue }) =>
buildManagementUrl(host, [
'bindings',
resolveVhost(vhost),
'e',
exchange,
destinationType === 'exchange' ? 'e' : 'q',
queue,
]),
method: 'POST',
headers: ({ username, password }) => buildAuthHeaders(username, password),
stripAuthOnRedirect: true,
body: ({ arguments: bindingArguments, routingKey }) => ({
routing_key: routingKey ?? '',
arguments: parseJsonObjectParam(bindingArguments, 'arguments') ?? {},
}),
},
transformResponse: async (response, params) => {
const exchange = params?.exchange ?? ''
const queueName = params?.queue ?? ''
const routingKey = params?.routingKey ?? ''
if (!response.ok) {
const error = await extractErrorMessage(response)
return {
success: false,
output: { exchange, queueName, routingKey, propertiesKey: null, created: false },
error,
}
}
return {
success: true,
output: {
exchange,
queueName,
routingKey,
propertiesKey: extractPropertiesKey(response.headers.get('location')),
created: true,
},
}
},
outputs: {
exchange: { type: 'string', description: 'Source exchange the binding reads from' },
queueName: { type: 'string', description: 'Destination queue the binding routes into' },
routingKey: { type: 'string', description: 'Routing key the binding matches' },
propertiesKey: {
type: 'string',
description: 'Broker identifier addressing the new binding',
optional: true,
},
created: { type: 'boolean', description: 'Whether the binding was created' },
},
}
+102
View File
@@ -0,0 +1,102 @@
import type {
RabbitmqCreateExchangeParams,
RabbitmqCreateExchangeResponse,
} from '@/tools/rabbitmq/types'
import {
buildAuthHeaders,
buildManagementUrl,
extractErrorMessage,
parseJsonObjectParam,
RABBITMQ_CONNECTION_PARAMS,
resolveVhost,
} from '@/tools/rabbitmq/utils'
import type { ToolConfig } from '@/tools/types'
const EXCHANGE_TYPES = new Set(['direct', 'fanout', 'topic', 'headers'])
export const rabbitmqCreateExchangeTool: ToolConfig<
RabbitmqCreateExchangeParams,
RabbitmqCreateExchangeResponse
> = {
id: 'rabbitmq_create_exchange',
name: 'RabbitMQ Create Exchange',
description:
'Declare a RabbitMQ exchange. Declaring an exchange that already exists with the same settings succeeds without changing it.',
version: '1.0.0',
params: {
...RABBITMQ_CONNECTION_PARAMS,
exchange: {
type: 'string',
required: true,
visibility: 'user-or-llm',
description: 'Name of the exchange to declare',
},
exchangeType: {
type: 'string',
required: false,
visibility: 'user-or-llm',
description:
'Routing behaviour: direct (exact routing key, default), topic (wildcard patterns), fanout (every bound queue), or headers (match on binding arguments)',
},
durable: {
type: 'boolean',
required: false,
visibility: 'user-only',
description: 'Whether the exchange survives a broker restart. Defaults to true',
},
autoDelete: {
type: 'boolean',
required: false,
visibility: 'user-only',
description: 'Delete the exchange once its last binding is removed. Defaults to false',
},
internal: {
type: 'boolean',
required: false,
visibility: 'user-only',
description:
'Internal exchanges cannot be published to directly, only bound from another exchange. Defaults to false',
},
arguments: {
type: 'string',
required: false,
visibility: 'user-or-llm',
description:
'Exchange arguments as a JSON object, e.g. {"alternate-exchange":"unrouted"} to capture messages that match no binding',
},
},
request: {
url: ({ host, vhost, exchange }) =>
buildManagementUrl(host, ['exchanges', resolveVhost(vhost), exchange]),
method: 'PUT',
headers: ({ username, password }) => buildAuthHeaders(username, password),
stripAuthOnRedirect: true,
body: ({ arguments: exchangeArguments, autoDelete, durable, exchangeType, internal }) => ({
type: EXCHANGE_TYPES.has(exchangeType ?? '') ? exchangeType : 'direct',
durable: durable !== false,
auto_delete: autoDelete === true,
internal: internal === true,
arguments: parseJsonObjectParam(exchangeArguments, 'arguments') ?? {},
}),
},
transformResponse: async (response, params) => {
const exchangeName = params?.exchange ?? ''
const vhost = params ? resolveVhost(params.vhost) : ''
if (!response.ok) {
const error = await extractErrorMessage(response)
return { success: false, output: { exchangeName, vhost, created: false }, error }
}
return { success: true, output: { exchangeName, vhost, created: true } }
},
outputs: {
exchangeName: { type: 'string', description: 'Name of the declared exchange' },
vhost: { type: 'string', description: 'Virtual host the exchange was declared in' },
created: { type: 'boolean', description: 'Whether the declaration succeeded' },
},
}
+103
View File
@@ -0,0 +1,103 @@
import type {
RabbitmqCreatePolicyParams,
RabbitmqCreatePolicyResponse,
} from '@/tools/rabbitmq/types'
import {
buildAuthHeaders,
buildManagementUrl,
extractErrorMessage,
parseJsonObjectParam,
RABBITMQ_CONNECTION_PARAMS,
resolveVhost,
} from '@/tools/rabbitmq/utils'
import type { ToolConfig } from '@/tools/types'
const APPLY_TO = new Set([
'all',
'queues',
'classic_queues',
'quorum_queues',
'streams',
'exchanges',
])
export const rabbitmqCreatePolicyTool: ToolConfig<
RabbitmqCreatePolicyParams,
RabbitmqCreatePolicyResponse
> = {
id: 'rabbitmq_create_policy',
name: 'RabbitMQ Create Policy',
description:
'Create or replace a RabbitMQ policy, applying settings such as dead-lettering, TTLs, or length limits to every queue or exchange whose name matches a pattern.',
version: '1.0.0',
params: {
...RABBITMQ_CONNECTION_PARAMS,
policyName: {
type: 'string',
required: true,
visibility: 'user-or-llm',
description: 'Name of the policy. Reusing an existing name replaces that policy',
},
pattern: {
type: 'string',
required: true,
visibility: 'user-or-llm',
description:
'Regular expression matched against queue or exchange names, e.g. ^orders\\. to match every name starting with orders.',
},
definition: {
type: 'string',
required: true,
visibility: 'user-or-llm',
description:
'Settings to apply, as a JSON object, e.g. {"dead-letter-exchange":"dlx","message-ttl":86400000,"max-length":10000}',
},
priority: {
type: 'number',
required: false,
visibility: 'user-or-llm',
description:
'Priority, defaulting to 0. When several policies match a resource only the highest-priority one applies — they do not merge',
},
applyTo: {
type: 'string',
required: false,
visibility: 'user-or-llm',
description:
'What the policy applies to: queues (default), classic_queues, quorum_queues, streams, exchanges, or all',
},
},
request: {
url: ({ host, vhost, policyName }) =>
buildManagementUrl(host, ['policies', resolveVhost(vhost), policyName]),
method: 'PUT',
headers: ({ username, password }) => buildAuthHeaders(username, password),
stripAuthOnRedirect: true,
body: ({ applyTo, definition, pattern, priority }) => ({
pattern: pattern,
definition: parseJsonObjectParam(definition, 'definition') ?? {},
priority: priority ?? 0,
'apply-to': APPLY_TO.has(applyTo ?? '') ? applyTo : 'queues',
}),
},
transformResponse: async (response, params) => {
const policyName = params?.policyName ?? ''
const vhost = params ? resolveVhost(params.vhost) : ''
if (!response.ok) {
const error = await extractErrorMessage(response)
return { success: false, output: { policyName, vhost, created: false }, error }
}
return { success: true, output: { policyName, vhost, created: true } }
},
outputs: {
policyName: { type: 'string', description: 'Name of the created policy' },
vhost: { type: 'string', description: 'Virtual host the policy applies in' },
created: { type: 'boolean', description: 'Whether the policy was created or replaced' },
},
}
+81
View File
@@ -0,0 +1,81 @@
import type { RabbitmqCreateQueueParams, RabbitmqCreateQueueResponse } from '@/tools/rabbitmq/types'
import {
buildAuthHeaders,
buildManagementUrl,
extractErrorMessage,
parseJsonObjectParam,
RABBITMQ_CONNECTION_PARAMS,
resolveVhost,
} from '@/tools/rabbitmq/utils'
import type { ToolConfig } from '@/tools/types'
export const rabbitmqCreateQueueTool: ToolConfig<
RabbitmqCreateQueueParams,
RabbitmqCreateQueueResponse
> = {
id: 'rabbitmq_create_queue',
name: 'RabbitMQ Create Queue',
description:
'Declare a RabbitMQ queue. Declaring a queue that already exists with the same settings succeeds without changing it.',
version: '1.0.0',
params: {
...RABBITMQ_CONNECTION_PARAMS,
queue: {
type: 'string',
required: true,
visibility: 'user-or-llm',
description: 'Name of the queue to declare',
},
durable: {
type: 'boolean',
required: false,
visibility: 'user-only',
description: 'Whether the queue survives a broker restart. Defaults to true',
},
autoDelete: {
type: 'boolean',
required: false,
visibility: 'user-only',
description: 'Delete the queue when its last consumer disconnects. Defaults to false',
},
arguments: {
type: 'string',
required: false,
visibility: 'user-or-llm',
description:
'Queue arguments as a JSON object, e.g. {"x-queue-type":"quorum","x-message-ttl":60000}',
},
},
request: {
url: ({ host, vhost, queue }) =>
buildManagementUrl(host, ['queues', resolveVhost(vhost), queue]),
method: 'PUT',
headers: ({ username, password }) => buildAuthHeaders(username, password),
stripAuthOnRedirect: true,
body: ({ arguments: queueArguments, autoDelete, durable }) => ({
durable: durable !== false,
auto_delete: autoDelete === true,
arguments: parseJsonObjectParam(queueArguments, 'arguments') ?? {},
}),
},
transformResponse: async (response, params) => {
const queueName = params?.queue ?? ''
const vhost = params ? resolveVhost(params.vhost) : ''
if (!response.ok) {
const error = await extractErrorMessage(response)
return { success: false, output: { queueName, vhost, created: false }, error }
}
return { success: true, output: { queueName, vhost, created: true } }
},
outputs: {
queueName: { type: 'string', description: 'Name of the declared queue' },
vhost: { type: 'string', description: 'Virtual host the queue was declared in' },
created: { type: 'boolean', description: 'Whether the declaration succeeded' },
},
}
+92
View File
@@ -0,0 +1,92 @@
import type {
RabbitmqDeleteBindingParams,
RabbitmqDeleteBindingResponse,
} from '@/tools/rabbitmq/types'
import {
buildAuthHeaders,
buildManagementUrl,
extractErrorMessage,
RABBITMQ_CONNECTION_PARAMS,
resolveVhost,
} from '@/tools/rabbitmq/utils'
import type { ToolConfig } from '@/tools/types'
export const rabbitmqDeleteBindingTool: ToolConfig<
RabbitmqDeleteBindingParams,
RabbitmqDeleteBindingResponse
> = {
id: 'rabbitmq_delete_binding',
name: 'RabbitMQ Delete Binding',
description:
'Remove a binding so an exchange stops routing its matching messages to that destination.',
version: '1.0.0',
params: {
...RABBITMQ_CONNECTION_PARAMS,
exchange: {
type: 'string',
required: true,
visibility: 'user-or-llm',
description: 'Source exchange the binding reads from',
},
destination: {
type: 'string',
required: true,
visibility: 'user-or-llm',
description: 'Destination queue or exchange the binding routes to',
},
destinationType: {
type: 'string',
required: false,
visibility: 'user-or-llm',
description: 'Whether the destination is a queue (default) or an exchange',
},
propertiesKey: {
type: 'string',
required: true,
visibility: 'user-or-llm',
description:
'Broker identifier for the binding, taken from List Bindings or Create Binding. It is the routing key for a simple binding, ~ for an empty routing key, and a hashed value when the binding has arguments',
},
},
request: {
url: ({ host, vhost, destination, destinationType, exchange, propertiesKey }) =>
buildManagementUrl(host, [
'bindings',
resolveVhost(vhost),
'e',
exchange,
destinationType === 'exchange' ? 'e' : 'q',
destination,
propertiesKey,
]),
method: 'DELETE',
headers: ({ username, password }) => buildAuthHeaders(username, password),
stripAuthOnRedirect: true,
},
transformResponse: async (response, params) => {
const exchange = params?.exchange ?? ''
const destination = params?.destination ?? ''
const propertiesKey = params?.propertiesKey ?? ''
if (!response.ok) {
const error = await extractErrorMessage(response)
return {
success: false,
output: { exchange, destination, propertiesKey, deleted: false },
error,
}
}
return { success: true, output: { exchange, destination, propertiesKey, deleted: true } }
},
outputs: {
exchange: { type: 'string', description: 'Source exchange the binding read from' },
destination: { type: 'string', description: 'Destination the binding routed to' },
propertiesKey: { type: 'string', description: 'Broker identifier of the deleted binding' },
deleted: { type: 'boolean', description: 'Whether the binding was deleted' },
},
}
@@ -0,0 +1,67 @@
import type {
RabbitmqDeleteExchangeParams,
RabbitmqDeleteExchangeResponse,
} from '@/tools/rabbitmq/types'
import {
buildAuthHeaders,
buildManagementUrl,
extractErrorMessage,
RABBITMQ_CONNECTION_PARAMS,
resolveVhost,
} from '@/tools/rabbitmq/utils'
import type { ToolConfig } from '@/tools/types'
export const rabbitmqDeleteExchangeTool: ToolConfig<
RabbitmqDeleteExchangeParams,
RabbitmqDeleteExchangeResponse
> = {
id: 'rabbitmq_delete_exchange',
name: 'RabbitMQ Delete Exchange',
description:
'Delete a RabbitMQ exchange and every binding attached to it. Publishers targeting it will fail afterwards.',
version: '1.0.0',
params: {
...RABBITMQ_CONNECTION_PARAMS,
exchange: {
type: 'string',
required: true,
visibility: 'user-or-llm',
description: 'Name of the exchange to delete',
},
ifUnused: {
type: 'boolean',
required: false,
visibility: 'user-only',
description: 'Only delete the exchange when nothing is bound to it',
},
},
request: {
url: ({ host, vhost, exchange, ifUnused }) =>
buildManagementUrl(host, ['exchanges', resolveVhost(vhost), exchange], {
'if-unused': ifUnused ? 'true' : undefined,
}),
method: 'DELETE',
headers: ({ username, password }) => buildAuthHeaders(username, password),
stripAuthOnRedirect: true,
},
transformResponse: async (response, params) => {
const exchangeName = params?.exchange ?? ''
const vhost = params ? resolveVhost(params.vhost) : ''
if (!response.ok) {
const error = await extractErrorMessage(response)
return { success: false, output: { exchangeName, vhost, deleted: false }, error }
}
return { success: true, output: { exchangeName, vhost, deleted: true } }
},
outputs: {
exchangeName: { type: 'string', description: 'Name of the deleted exchange' },
vhost: { type: 'string', description: 'Virtual host the exchange was deleted from' },
deleted: { type: 'boolean', description: 'Whether the exchange was deleted' },
},
}
+59
View File
@@ -0,0 +1,59 @@
import type {
RabbitmqDeletePolicyParams,
RabbitmqDeletePolicyResponse,
} from '@/tools/rabbitmq/types'
import {
buildAuthHeaders,
buildManagementUrl,
extractErrorMessage,
RABBITMQ_CONNECTION_PARAMS,
resolveVhost,
} from '@/tools/rabbitmq/utils'
import type { ToolConfig } from '@/tools/types'
export const rabbitmqDeletePolicyTool: ToolConfig<
RabbitmqDeletePolicyParams,
RabbitmqDeletePolicyResponse
> = {
id: 'rabbitmq_delete_policy',
name: 'RabbitMQ Delete Policy',
description:
'Delete a RabbitMQ policy. Every queue and exchange it matched immediately loses the settings it applied.',
version: '1.0.0',
params: {
...RABBITMQ_CONNECTION_PARAMS,
policyName: {
type: 'string',
required: true,
visibility: 'user-or-llm',
description: 'Name of the policy to delete',
},
},
request: {
url: ({ host, vhost, policyName }) =>
buildManagementUrl(host, ['policies', resolveVhost(vhost), policyName]),
method: 'DELETE',
headers: ({ username, password }) => buildAuthHeaders(username, password),
stripAuthOnRedirect: true,
},
transformResponse: async (response, params) => {
const policyName = params?.policyName ?? ''
const vhost = params ? resolveVhost(params.vhost) : ''
if (!response.ok) {
const error = await extractErrorMessage(response)
return { success: false, output: { policyName, vhost, deleted: false }, error }
}
return { success: true, output: { policyName, vhost, deleted: true } }
},
outputs: {
policyName: { type: 'string', description: 'Name of the deleted policy' },
vhost: { type: 'string', description: 'Virtual host the policy applied in' },
deleted: { type: 'boolean', description: 'Whether the policy was deleted' },
},
}
+71
View File
@@ -0,0 +1,71 @@
import type { RabbitmqDeleteQueueParams, RabbitmqDeleteQueueResponse } from '@/tools/rabbitmq/types'
import {
buildAuthHeaders,
buildManagementUrl,
extractErrorMessage,
RABBITMQ_CONNECTION_PARAMS,
resolveVhost,
} from '@/tools/rabbitmq/utils'
import type { ToolConfig } from '@/tools/types'
export const rabbitmqDeleteQueueTool: ToolConfig<
RabbitmqDeleteQueueParams,
RabbitmqDeleteQueueResponse
> = {
id: 'rabbitmq_delete_queue',
name: 'RabbitMQ Delete Queue',
description:
'Delete a RabbitMQ queue and every message still in it. Can be guarded so the delete only happens when the queue is unused or empty.',
version: '1.0.0',
params: {
...RABBITMQ_CONNECTION_PARAMS,
queue: {
type: 'string',
required: true,
visibility: 'user-or-llm',
description: 'Name of the queue to delete',
},
ifUnused: {
type: 'boolean',
required: false,
visibility: 'user-only',
description: 'Only delete the queue when it has no consumers',
},
ifEmpty: {
type: 'boolean',
required: false,
visibility: 'user-only',
description: 'Only delete the queue when it holds no messages',
},
},
request: {
url: ({ host, vhost, ifEmpty, ifUnused, queue }) =>
buildManagementUrl(host, ['queues', resolveVhost(vhost), queue], {
'if-unused': ifUnused ? 'true' : undefined,
'if-empty': ifEmpty ? 'true' : undefined,
}),
method: 'DELETE',
headers: ({ username, password }) => buildAuthHeaders(username, password),
stripAuthOnRedirect: true,
},
transformResponse: async (response, params) => {
const queueName = params?.queue ?? ''
const vhost = params ? resolveVhost(params.vhost) : ''
if (!response.ok) {
const error = await extractErrorMessage(response)
return { success: false, output: { queueName, vhost, deleted: false }, error }
}
return { success: true, output: { queueName, vhost, deleted: true } }
},
outputs: {
queueName: { type: 'string', description: 'Name of the deleted queue' },
vhost: { type: 'string', description: 'Virtual host the queue was deleted from' },
deleted: { type: 'boolean', description: 'Whether the queue was deleted' },
},
}
+59
View File
@@ -0,0 +1,59 @@
import type { RabbitmqGetExchangeParams, RabbitmqGetExchangeResponse } from '@/tools/rabbitmq/types'
import {
buildAuthHeaders,
buildManagementUrl,
extractErrorMessage,
projectExchange,
RABBITMQ_CONNECTION_PARAMS,
RABBITMQ_EXCHANGE_OUTPUT_PROPERTIES,
resolveVhost,
} from '@/tools/rabbitmq/utils'
import type { ToolConfig } from '@/tools/types'
export const rabbitmqGetExchangeTool: ToolConfig<
RabbitmqGetExchangeParams,
RabbitmqGetExchangeResponse
> = {
id: 'rabbitmq_get_exchange',
name: 'RabbitMQ Get Exchange',
description: 'Read a single RabbitMQ exchange and the settings it was declared with.',
version: '1.0.0',
params: {
...RABBITMQ_CONNECTION_PARAMS,
exchange: {
type: 'string',
required: false,
visibility: 'user-or-llm',
description:
'Exchange name to read. Leave empty for the default exchange, which is a valid value, so this is not required',
},
},
request: {
url: ({ host, vhost, exchange }) =>
buildManagementUrl(host, ['exchanges', resolveVhost(vhost), exchange ?? '']),
method: 'GET',
headers: ({ username, password }) => buildAuthHeaders(username, password),
stripAuthOnRedirect: true,
},
transformResponse: async (response) => {
if (!response.ok) {
const error = await extractErrorMessage(response)
return { success: false, output: { exchange: projectExchange(null) }, error }
}
const data = await response.json()
return { success: true, output: { exchange: projectExchange(data) } }
},
outputs: {
exchange: {
type: 'object',
description: 'The requested exchange',
properties: RABBITMQ_EXCHANGE_OUTPUT_PROPERTIES,
},
},
}
+165
View File
@@ -0,0 +1,165 @@
import type { RabbitmqGetMessagesParams, RabbitmqGetMessagesResponse } from '@/tools/rabbitmq/types'
import {
buildAuthHeaders,
buildManagementUrl,
extractErrorMessage,
projectMessage,
RABBITMQ_CONNECTION_PARAMS,
RABBITMQ_MESSAGE_OUTPUT_PROPERTIES,
resolveVhost,
} from '@/tools/rabbitmq/utils'
import type { ToolConfig } from '@/tools/types'
const ACK_MODES = new Set([
'ack_requeue_true',
'ack_requeue_false',
'reject_requeue_true',
'reject_requeue_false',
])
/**
* The broker applies no upper bound of its own to `count`, so a single call could pull an
* unbounded number of messages into memory, and the shared tool transport rejects any response
* body over 10MB before a tool ever sees it. Both dimensions are bounded here.
*/
const MAX_MESSAGE_COUNT = 50
const DEFAULT_TRUNCATE_BYTES = 50_000
const MIN_TRUNCATE_BYTES = 1_024
const MAX_TRUNCATE_BYTES = 1_000_000
/**
* Budget for one retrieval, kept under the transport's 10MB cap with room for the JSON envelope.
*/
const RESPONSE_BUDGET_BYTES = 8_000_000
/**
* Per-message allowance for AMQP properties and headers. `truncate` bounds the payload only —
* the broker returns properties in full — so the budget has to reserve space for them rather
* than assume they are small. This figure is RabbitMQ's default `frame_max`, which bounds the
* content header frame carrying a message's properties, so any message published by a standard
* AMQP client fits inside it.
*
* The one case this cannot cover is a message published through the management HTTP API itself,
* which bypasses `frame_max` and accepts properties up to that API's ~10MB request-body limit.
* Retrieving several of those can still exceed the transport cap and surfaces as a response-size
* error; the remedy is a lower `count`.
*/
const PER_MESSAGE_METADATA_RESERVE_BYTES = 131_072
/** base64 payloads inflate 4/3 on the wire before JSON escaping. */
const BASE64_INFLATION = 4 / 3
function resolveCount(count: number | undefined): number {
if (typeof count !== 'number' || !Number.isFinite(count)) return 1
return Math.min(Math.max(Math.trunc(count), 1), MAX_MESSAGE_COUNT)
}
/**
* Resolves the per-message payload limit sent to the broker. The whole batch — payloads plus the
* reserved metadata allowance for every message — is held inside {@link RESPONSE_BUDGET_BYTES},
* so asking for a large `truncate` alongside a large `count` yields shorter payloads rather than
* a retrieval that fails at the transport cap.
*/
function resolveTruncate(truncate: number | undefined, count: number | undefined): number {
const messages = resolveCount(count)
const requested =
typeof truncate === 'number' && Number.isFinite(truncate)
? Math.max(Math.trunc(truncate), 1)
: DEFAULT_TRUNCATE_BYTES
const payloadBudget = RESPONSE_BUDGET_BYTES - messages * PER_MESSAGE_METADATA_RESERVE_BYTES
const perMessageBudget = Math.floor(payloadBudget / messages / BASE64_INFLATION)
return Math.max(Math.min(requested, MAX_TRUNCATE_BYTES, perMessageBudget), MIN_TRUNCATE_BYTES)
}
export const rabbitmqGetMessagesTool: ToolConfig<
RabbitmqGetMessagesParams,
RabbitmqGetMessagesResponse
> = {
id: 'rabbitmq_get_messages',
name: 'RabbitMQ Get Messages',
description:
'Retrieve messages from a RabbitMQ queue. Defaults to requeueing the messages so they stay available to real consumers.',
version: '1.0.0',
params: {
...RABBITMQ_CONNECTION_PARAMS,
queue: {
type: 'string',
required: true,
visibility: 'user-or-llm',
description: 'Queue to read messages from',
},
count: {
type: 'number',
required: false,
visibility: 'user-or-llm',
description: `Maximum number of messages to retrieve, from 1 to ${MAX_MESSAGE_COUNT}. Defaults to 1`,
},
ackmode: {
type: 'string',
required: false,
visibility: 'user-only',
description:
'How retrieved messages are handled: ack_requeue_true (default, leaves messages in the queue), ack_requeue_false (removes them), reject_requeue_true, or reject_requeue_false',
},
encoding: {
type: 'string',
required: false,
visibility: 'user-only',
description:
'auto (default) returns readable text where possible, base64 always returns base64',
},
truncate: {
type: 'number',
required: false,
visibility: 'user-only',
description: `Truncate payloads longer than this many bytes. Defaults to ${DEFAULT_TRUNCATE_BYTES}, capped at ${MAX_TRUNCATE_BYTES}, and lowered further at high counts so the whole batch stays inside the response limit. Each message reports whether it was truncated`,
},
},
request: {
url: ({ host, vhost, queue }) =>
buildManagementUrl(host, ['queues', resolveVhost(vhost), queue, 'get']),
method: 'POST',
headers: ({ username, password }) => buildAuthHeaders(username, password),
stripAuthOnRedirect: true,
body: ({ ackmode, count, encoding, truncate }) => ({
count: resolveCount(count),
ackmode: ACK_MODES.has(ackmode ?? '') ? ackmode : 'ack_requeue_true',
encoding: encoding === 'base64' ? 'base64' : 'auto',
truncate: resolveTruncate(truncate, count),
}),
},
transformResponse: async (response, params) => {
const queueName = params?.queue ?? ''
if (!response.ok) {
const error = await extractErrorMessage(response)
return { success: false, output: { queueName, count: 0, messages: [] }, error }
}
const data = await response.json()
const truncateLimit = resolveTruncate(params?.truncate, params?.count)
const messages = Array.isArray(data)
? data.map((message) => projectMessage(message, truncateLimit))
: []
return {
success: true,
output: { queueName, count: messages.length, messages },
}
},
outputs: {
queueName: { type: 'string', description: 'Queue the messages were read from' },
count: { type: 'number', description: 'Number of messages retrieved' },
messages: {
type: 'array',
description: 'Retrieved messages, empty when the queue holds nothing',
items: { type: 'object', properties: RABBITMQ_MESSAGE_OUTPUT_PROPERTIES },
},
},
}
+110
View File
@@ -0,0 +1,110 @@
import type { RabbitmqGetOverviewParams, RabbitmqGetOverviewResponse } from '@/tools/rabbitmq/types'
import {
buildAuthHeaders,
buildManagementUrl,
extractErrorMessage,
RABBITMQ_CONNECTION_PARAMS,
} from '@/tools/rabbitmq/utils'
import type { ToolConfig } from '@/tools/types'
const EMPTY_OVERVIEW = {
rabbitmqVersion: null,
productName: null,
productVersion: null,
erlangVersion: null,
clusterName: null,
node: null,
objectTotals: {},
queueTotals: {},
messageStats: {},
} as const
function asRecord(value: unknown): Record<string, unknown> {
return typeof value === 'object' && value !== null && !Array.isArray(value)
? (value as Record<string, unknown>)
: {}
}
function asStringOrNull(value: unknown): string | null {
return typeof value === 'string' ? value : null
}
export const rabbitmqGetOverviewTool: ToolConfig<
RabbitmqGetOverviewParams,
RabbitmqGetOverviewResponse
> = {
id: 'rabbitmq_get_overview',
name: 'RabbitMQ Get Overview',
description:
'Read broker-wide RabbitMQ status: version, cluster name, object totals, queue depth totals, and message rates.',
version: '1.0.0',
params: { ...RABBITMQ_CONNECTION_PARAMS },
request: {
url: ({ host }) => buildManagementUrl(host, ['overview']),
method: 'GET',
headers: ({ username, password }) => buildAuthHeaders(username, password),
stripAuthOnRedirect: true,
},
transformResponse: async (response) => {
if (!response.ok) {
const error = await extractErrorMessage(response)
return { success: false, output: { ...EMPTY_OVERVIEW }, error }
}
const data = await response.json()
return {
success: true,
output: {
rabbitmqVersion: asStringOrNull(data?.rabbitmq_version),
productName: asStringOrNull(data?.product_name),
productVersion: asStringOrNull(data?.product_version),
erlangVersion: asStringOrNull(data?.erlang_version),
clusterName: asStringOrNull(data?.cluster_name),
node: asStringOrNull(data?.node),
objectTotals: asRecord(data?.object_totals),
queueTotals: asRecord(data?.queue_totals),
messageStats: asRecord(data?.message_stats),
},
}
},
outputs: {
rabbitmqVersion: { type: 'string', description: 'RabbitMQ version running on the node' },
productName: { type: 'string', description: 'Broker product name' },
productVersion: { type: 'string', description: 'Broker product version' },
erlangVersion: { type: 'string', description: 'Erlang runtime version' },
clusterName: { type: 'string', description: 'Name of the cluster' },
node: { type: 'string', description: 'Node that served the request' },
objectTotals: {
type: 'object',
description: 'Counts of brokers objects',
properties: {
connections: { type: 'number', description: 'Open connections' },
channels: { type: 'number', description: 'Open channels' },
exchanges: { type: 'number', description: 'Declared exchanges' },
queues: { type: 'number', description: 'Declared queues' },
consumers: { type: 'number', description: 'Registered consumers' },
},
},
queueTotals: {
type: 'object',
description: 'Aggregate queue depth across the broker',
properties: {
messages: { type: 'number', description: 'Total messages across all queues' },
messages_ready: { type: 'number', description: 'Messages ready for delivery' },
messages_unacknowledged: {
type: 'number',
description: 'Delivered but unacknowledged messages',
},
},
},
messageStats: {
type: 'json',
description: 'Broker-wide message counters and rates, e.g. publish and confirm totals',
},
},
}
+56
View File
@@ -0,0 +1,56 @@
import type { RabbitmqGetQueueParams, RabbitmqGetQueueResponse } from '@/tools/rabbitmq/types'
import {
buildAuthHeaders,
buildManagementUrl,
extractErrorMessage,
projectQueue,
RABBITMQ_CONNECTION_PARAMS,
RABBITMQ_QUEUE_OUTPUT_PROPERTIES,
resolveVhost,
} from '@/tools/rabbitmq/utils'
import type { ToolConfig } from '@/tools/types'
export const rabbitmqGetQueueTool: ToolConfig<RabbitmqGetQueueParams, RabbitmqGetQueueResponse> = {
id: 'rabbitmq_get_queue',
name: 'RabbitMQ Get Queue',
description:
'Read a single RabbitMQ queue, including its depth, consumer count, and declaration settings.',
version: '1.0.0',
params: {
...RABBITMQ_CONNECTION_PARAMS,
queue: {
type: 'string',
required: true,
visibility: 'user-or-llm',
description: 'Queue name to read',
},
},
request: {
url: ({ host, vhost, queue }) =>
buildManagementUrl(host, ['queues', resolveVhost(vhost), queue]),
method: 'GET',
headers: ({ username, password }) => buildAuthHeaders(username, password),
stripAuthOnRedirect: true,
},
transformResponse: async (response) => {
if (!response.ok) {
const error = await extractErrorMessage(response)
return { success: false, output: { queue: projectQueue(null) }, error }
}
const data = await response.json()
return { success: true, output: { queue: projectQueue(data) } }
},
outputs: {
queue: {
type: 'object',
description: 'The requested queue',
properties: RABBITMQ_QUEUE_OUTPUT_PROPERTIES,
},
},
}
+164
View File
@@ -0,0 +1,164 @@
import type { RabbitmqHealthCheckParams, RabbitmqHealthCheckResponse } from '@/tools/rabbitmq/types'
import {
buildAuthHeaders,
buildManagementUrl,
extractErrorMessage,
RABBITMQ_CONNECTION_PARAMS,
} from '@/tools/rabbitmq/utils'
import type { ToolConfig } from '@/tools/types'
const CHECKS = new Set([
'alarms',
'local-alarms',
'virtual-hosts',
'node-is-quorum-critical',
'port-listener',
'protocol-listener',
'certificate-expiration',
])
const EXPIRY_UNITS = new Set(['days', 'weeks', 'months', 'years'])
function resolveCheck(check: string | undefined): string {
return CHECKS.has(check ?? '') ? (check as string) : 'alarms'
}
/** Builds the trailing path segments for the checks that take arguments. */
function checkSegments({
check: requestedCheck,
port,
protocol,
within,
unit,
}: Omit<RabbitmqHealthCheckParams, 'host' | 'username' | 'password' | 'vhost'>): string[] {
const check = resolveCheck(requestedCheck)
if (check === 'port-listener') {
if (port === undefined) {
throw new Error('port is required for the port-listener health check')
}
return [check, String(port)]
}
if (check === 'protocol-listener') {
if (!protocol?.trim()) {
throw new Error('protocol is required for the protocol-listener health check')
}
return [check, protocol]
}
if (check === 'certificate-expiration') {
if (within === undefined) {
throw new Error('within is required for the certificate-expiration health check')
}
return [check, String(within), EXPIRY_UNITS.has(unit ?? '') ? (unit as string) : 'months']
}
return [check]
}
export const rabbitmqHealthCheckTool: ToolConfig<
RabbitmqHealthCheckParams,
RabbitmqHealthCheckResponse
> = {
id: 'rabbitmq_health_check',
name: 'RabbitMQ Health Check',
description:
'Run one of the broker health checks and report whether it passed. A failing check is a normal result, not a tool error.',
version: '1.0.0',
params: {
...RABBITMQ_CONNECTION_PARAMS,
check: {
type: 'string',
required: false,
visibility: 'user-or-llm',
description:
'Which check to run: alarms (cluster-wide resource alarms, default), local-alarms, virtual-hosts, node-is-quorum-critical, port-listener, protocol-listener, or certificate-expiration',
},
port: {
type: 'number',
required: false,
visibility: 'user-or-llm',
description: 'Port to verify a listener on. Required for the port-listener check',
},
protocol: {
type: 'string',
required: false,
visibility: 'user-or-llm',
description:
'Protocol to verify a listener for, e.g. amqp, amqp/ssl, mqtt, stomp, or http. Required for the protocol-listener check',
},
within: {
type: 'number',
required: false,
visibility: 'user-or-llm',
description:
'How far ahead to look for expiring certificates. Required for the certificate-expiration check',
},
unit: {
type: 'string',
required: false,
visibility: 'user-or-llm',
description:
'Unit for the certificate-expiration window: days, weeks, months (default), or years',
},
},
request: {
url: ({ host, ...rest }) =>
buildManagementUrl(host, ['health', 'checks', ...checkSegments(rest)]),
method: 'GET',
headers: ({ username, password }) => buildAuthHeaders(username, password),
stripAuthOnRedirect: true,
},
/**
* A failed health check answers with HTTP 503 and a body explaining why. That is a meaningful
* answer to the question the caller asked, so it is reported as a successful call with
* `healthy: false` rather than a tool error. Genuine transport and auth failures — anything
* that is not 200 or 503 — still surface as errors.
*/
transformResponse: async (response, params) => {
const check = resolveCheck(params?.check)
if (!response.ok && response.status !== 503) {
const error = await extractErrorMessage(response)
return {
success: false,
output: { check, healthy: false, status: 'unknown', reason: null, details: {} },
error,
}
}
const data = await response.json()
const status = typeof data?.status === 'string' ? data.status : 'unknown'
return {
success: true,
output: {
check,
healthy: status === 'ok',
status,
reason: typeof data?.reason === 'string' ? data.reason : null,
details: typeof data === 'object' && data !== null ? data : {},
},
}
},
outputs: {
check: { type: 'string', description: 'The health check that was run' },
healthy: { type: 'boolean', description: 'True when the check reported status ok' },
status: { type: 'string', description: 'Raw status reported by the broker: ok or failed' },
reason: {
type: 'string',
description: 'Explanation the broker gave, present on failures and on some passes',
optional: true,
},
details: {
type: 'json',
description:
'Full check body, including check-specific fields such as the ports or protocols found',
},
},
}
+25
View File
@@ -0,0 +1,25 @@
export { rabbitmqCreateBindingTool } from '@/tools/rabbitmq/create_binding'
export { rabbitmqCreateExchangeTool } from '@/tools/rabbitmq/create_exchange'
export { rabbitmqCreatePolicyTool } from '@/tools/rabbitmq/create_policy'
export { rabbitmqCreateQueueTool } from '@/tools/rabbitmq/create_queue'
export { rabbitmqDeleteBindingTool } from '@/tools/rabbitmq/delete_binding'
export { rabbitmqDeleteExchangeTool } from '@/tools/rabbitmq/delete_exchange'
export { rabbitmqDeletePolicyTool } from '@/tools/rabbitmq/delete_policy'
export { rabbitmqDeleteQueueTool } from '@/tools/rabbitmq/delete_queue'
export { rabbitmqGetExchangeTool } from '@/tools/rabbitmq/get_exchange'
export { rabbitmqGetMessagesTool } from '@/tools/rabbitmq/get_messages'
export { rabbitmqGetOverviewTool } from '@/tools/rabbitmq/get_overview'
export { rabbitmqGetQueueTool } from '@/tools/rabbitmq/get_queue'
export { rabbitmqHealthCheckTool } from '@/tools/rabbitmq/health_check'
export { rabbitmqListBindingsTool } from '@/tools/rabbitmq/list_bindings'
export { rabbitmqListChannelsTool } from '@/tools/rabbitmq/list_channels'
export { rabbitmqListConnectionsTool } from '@/tools/rabbitmq/list_connections'
export { rabbitmqListConsumersTool } from '@/tools/rabbitmq/list_consumers'
export { rabbitmqListExchangeBindingsTool } from '@/tools/rabbitmq/list_exchange_bindings'
export { rabbitmqListExchangesTool } from '@/tools/rabbitmq/list_exchanges'
export { rabbitmqListNodesTool } from '@/tools/rabbitmq/list_nodes'
export { rabbitmqListPoliciesTool } from '@/tools/rabbitmq/list_policies'
export { rabbitmqListQueuesTool } from '@/tools/rabbitmq/list_queues'
export { rabbitmqListVhostsTool } from '@/tools/rabbitmq/list_vhosts'
export { rabbitmqPublishMessageTool } from '@/tools/rabbitmq/publish_message'
export { rabbitmqPurgeQueueTool } from '@/tools/rabbitmq/purge_queue'
+68
View File
@@ -0,0 +1,68 @@
import type {
RabbitmqListBindingsParams,
RabbitmqListBindingsResponse,
} from '@/tools/rabbitmq/types'
import {
buildAuthHeaders,
buildManagementUrl,
extractErrorMessage,
projectBinding,
RABBITMQ_BINDING_OUTPUT_PROPERTIES,
RABBITMQ_CONNECTION_PARAMS,
resolveVhost,
} from '@/tools/rabbitmq/utils'
import type { ToolConfig } from '@/tools/types'
export const rabbitmqListBindingsTool: ToolConfig<
RabbitmqListBindingsParams,
RabbitmqListBindingsResponse
> = {
id: 'rabbitmq_list_bindings',
name: 'RabbitMQ List Bindings',
description:
'List the bindings that route messages into a RabbitMQ queue, including the implicit default-exchange binding.',
version: '1.0.0',
params: {
...RABBITMQ_CONNECTION_PARAMS,
queue: {
type: 'string',
required: true,
visibility: 'user-or-llm',
description: 'Queue whose bindings should be listed',
},
},
request: {
url: ({ host, vhost, queue }) =>
buildManagementUrl(host, ['queues', resolveVhost(vhost), queue, 'bindings']),
method: 'GET',
headers: ({ username, password }) => buildAuthHeaders(username, password),
stripAuthOnRedirect: true,
},
transformResponse: async (response, params) => {
const queueName = params?.queue ?? ''
if (!response.ok) {
const error = await extractErrorMessage(response)
return { success: false, output: { queueName, bindings: [], count: 0 }, error }
}
const data = await response.json()
const bindings = Array.isArray(data) ? data.map(projectBinding) : []
return { success: true, output: { queueName, bindings, count: bindings.length } }
},
outputs: {
queueName: { type: 'string', description: 'Queue the bindings route into' },
bindings: {
type: 'array',
description:
'Bindings targeting the queue. The entry with an empty source is the implicit default-exchange binding',
items: { type: 'object', properties: RABBITMQ_BINDING_OUTPUT_PROPERTIES },
},
count: { type: 'number', description: 'Number of bindings returned' },
},
}
+105
View File
@@ -0,0 +1,105 @@
import type {
RabbitmqChannel,
RabbitmqListChannelsParams,
RabbitmqListChannelsResponse,
} from '@/tools/rabbitmq/types'
import {
buildAuthHeaders,
buildManagementUrl,
clampPageSize,
extractErrorMessage,
projectChannel,
RABBITMQ_CHANNEL_COLUMNS,
RABBITMQ_CHANNEL_OUTPUT_PROPERTIES,
RABBITMQ_CONNECTION_PARAMS,
RABBITMQ_MAX_PAGE_SIZE,
unwrapPaginated,
} from '@/tools/rabbitmq/utils'
import type { ToolConfig } from '@/tools/types'
const DEFAULT_PAGE_SIZE = 50
export const rabbitmqListChannelsTool: ToolConfig<
RabbitmqListChannelsParams,
RabbitmqListChannelsResponse
> = {
id: 'rabbitmq_list_channels',
name: 'RabbitMQ List Channels',
description:
'List open channels with their prefetch limit and unacknowledged message count, which is where stalled consumers show up. Channels are cluster-wide, not scoped to one virtual host.',
version: '1.0.0',
params: {
...RABBITMQ_CONNECTION_PARAMS,
page: {
type: 'number',
required: false,
visibility: 'user-or-llm',
description: 'Page of results to return, starting at 1',
},
pageSize: {
type: 'number',
required: false,
visibility: 'user-or-llm',
description: `Channels per page, from 1 to ${RABBITMQ_MAX_PAGE_SIZE}. Defaults to ${DEFAULT_PAGE_SIZE}`,
},
name: {
type: 'string',
required: false,
visibility: 'user-or-llm',
description: 'Filter channels whose name contains this value',
},
useRegex: {
type: 'boolean',
required: false,
visibility: 'user-only',
description: 'Treat the name filter as a regular expression',
},
},
request: {
url: ({ host, name, page, pageSize, useRegex }) =>
buildManagementUrl(host, ['channels'], {
page: page ?? 1,
page_size: clampPageSize(pageSize, DEFAULT_PAGE_SIZE),
name: name,
use_regex: useRegex ? 'true' : undefined,
columns: RABBITMQ_CHANNEL_COLUMNS,
}),
method: 'GET',
headers: ({ username, password }) => buildAuthHeaders(username, password),
stripAuthOnRedirect: true,
},
transformResponse: async (response) => {
if (!response.ok) {
const error = await extractErrorMessage(response)
return {
success: false,
output: { channels: [], count: 0, totalCount: null, page: null, pageCount: null },
error,
}
}
const data = await response.json()
const { items, totalCount, page, pageCount } = unwrapPaginated<RabbitmqChannel>(data)
const channels = items.map(projectChannel)
return {
success: true,
output: { channels, count: channels.length, totalCount, page, pageCount },
}
},
outputs: {
channels: {
type: 'array',
description: 'Open channels',
items: { type: 'object', properties: RABBITMQ_CHANNEL_OUTPUT_PROPERTIES },
},
count: { type: 'number', description: 'Number of channels returned on this page' },
totalCount: { type: 'number', description: 'Total channels before filtering', optional: true },
page: { type: 'number', description: 'Page number returned', optional: true },
pageCount: { type: 'number', description: 'Total number of pages', optional: true },
},
}
+109
View File
@@ -0,0 +1,109 @@
import type {
RabbitmqConnection,
RabbitmqListConnectionsParams,
RabbitmqListConnectionsResponse,
} from '@/tools/rabbitmq/types'
import {
buildAuthHeaders,
buildManagementUrl,
clampPageSize,
extractErrorMessage,
projectConnection,
RABBITMQ_CONNECTION_COLUMNS,
RABBITMQ_CONNECTION_OUTPUT_PROPERTIES,
RABBITMQ_CONNECTION_PARAMS,
RABBITMQ_MAX_PAGE_SIZE,
unwrapPaginated,
} from '@/tools/rabbitmq/utils'
import type { ToolConfig } from '@/tools/types'
const DEFAULT_PAGE_SIZE = 50
export const rabbitmqListConnectionsTool: ToolConfig<
RabbitmqListConnectionsParams,
RabbitmqListConnectionsResponse
> = {
id: 'rabbitmq_list_connections',
name: 'RabbitMQ List Connections',
description:
'List client connections to the broker with their user, state, and channel count. Connections are cluster-wide, not scoped to one virtual host.',
version: '1.0.0',
params: {
...RABBITMQ_CONNECTION_PARAMS,
page: {
type: 'number',
required: false,
visibility: 'user-or-llm',
description: 'Page of results to return, starting at 1',
},
pageSize: {
type: 'number',
required: false,
visibility: 'user-or-llm',
description: `Connections per page, from 1 to ${RABBITMQ_MAX_PAGE_SIZE}. Defaults to ${DEFAULT_PAGE_SIZE}`,
},
name: {
type: 'string',
required: false,
visibility: 'user-or-llm',
description: 'Filter connections whose name contains this value',
},
useRegex: {
type: 'boolean',
required: false,
visibility: 'user-only',
description: 'Treat the name filter as a regular expression',
},
},
request: {
url: ({ host, name, page, pageSize, useRegex }) =>
buildManagementUrl(host, ['connections'], {
page: page ?? 1,
page_size: clampPageSize(pageSize, DEFAULT_PAGE_SIZE),
name: name,
use_regex: useRegex ? 'true' : undefined,
columns: RABBITMQ_CONNECTION_COLUMNS,
}),
method: 'GET',
headers: ({ username, password }) => buildAuthHeaders(username, password),
stripAuthOnRedirect: true,
},
transformResponse: async (response) => {
if (!response.ok) {
const error = await extractErrorMessage(response)
return {
success: false,
output: { connections: [], count: 0, totalCount: null, page: null, pageCount: null },
error,
}
}
const data = await response.json()
const { items, totalCount, page, pageCount } = unwrapPaginated<RabbitmqConnection>(data)
const connections = items.map(projectConnection)
return {
success: true,
output: { connections, count: connections.length, totalCount, page, pageCount },
}
},
outputs: {
connections: {
type: 'array',
description: 'Open client connections',
items: { type: 'object', properties: RABBITMQ_CONNECTION_OUTPUT_PROPERTIES },
},
count: { type: 'number', description: 'Number of connections returned on this page' },
totalCount: {
type: 'number',
description: 'Total connections before filtering',
optional: true,
},
page: { type: 'number', description: 'Page number returned', optional: true },
pageCount: { type: 'number', description: 'Total number of pages', optional: true },
},
}
+59
View File
@@ -0,0 +1,59 @@
import type {
RabbitmqListConsumersParams,
RabbitmqListConsumersResponse,
} from '@/tools/rabbitmq/types'
import {
buildAuthHeaders,
buildManagementUrl,
extractErrorMessage,
projectConsumer,
RABBITMQ_CONNECTION_PARAMS,
RABBITMQ_CONSUMER_COLUMNS,
RABBITMQ_CONSUMER_OUTPUT_PROPERTIES,
resolveVhost,
} from '@/tools/rabbitmq/utils'
import type { ToolConfig } from '@/tools/types'
export const rabbitmqListConsumersTool: ToolConfig<
RabbitmqListConsumersParams,
RabbitmqListConsumersResponse
> = {
id: 'rabbitmq_list_consumers',
name: 'RabbitMQ List Consumers',
description:
'List the consumers subscribed in a virtual host. An empty result for a queue with a backlog means nothing is processing it.',
version: '1.0.0',
params: { ...RABBITMQ_CONNECTION_PARAMS },
request: {
url: ({ host, vhost }) =>
buildManagementUrl(host, ['consumers', resolveVhost(vhost)], {
columns: RABBITMQ_CONSUMER_COLUMNS,
}),
method: 'GET',
headers: ({ username, password }) => buildAuthHeaders(username, password),
stripAuthOnRedirect: true,
},
transformResponse: async (response) => {
if (!response.ok) {
const error = await extractErrorMessage(response)
return { success: false, output: { consumers: [], count: 0 }, error }
}
const data = await response.json()
const consumers = Array.isArray(data) ? data.map(projectConsumer) : []
return { success: true, output: { consumers, count: consumers.length } }
},
outputs: {
consumers: {
type: 'array',
description: 'Consumers currently subscribed in the virtual host',
items: { type: 'object', properties: RABBITMQ_CONSUMER_OUTPUT_PROPERTIES },
},
count: { type: 'number', description: 'Number of consumers returned' },
},
}
@@ -0,0 +1,75 @@
import type {
RabbitmqListExchangeBindingsParams,
RabbitmqListExchangeBindingsResponse,
} from '@/tools/rabbitmq/types'
import {
buildAuthHeaders,
buildManagementUrl,
extractErrorMessage,
projectBinding,
RABBITMQ_BINDING_OUTPUT_PROPERTIES,
RABBITMQ_CONNECTION_PARAMS,
resolveVhost,
} from '@/tools/rabbitmq/utils'
import type { ToolConfig } from '@/tools/types'
export const rabbitmqListExchangeBindingsTool: ToolConfig<
RabbitmqListExchangeBindingsParams,
RabbitmqListExchangeBindingsResponse
> = {
id: 'rabbitmq_list_exchange_bindings',
name: 'RabbitMQ List Exchange Bindings',
description:
'List everything an exchange routes to, so you can see which routing keys reach which queues.',
version: '1.0.0',
params: {
...RABBITMQ_CONNECTION_PARAMS,
exchange: {
type: 'string',
required: false,
visibility: 'user-or-llm',
description:
'Exchange whose outgoing bindings should be listed. Leave empty for the default exchange, which is a valid value, so this is not required',
},
},
request: {
url: ({ host, vhost, exchange }) =>
buildManagementUrl(host, [
'exchanges',
resolveVhost(vhost),
exchange ?? '',
'bindings',
'source',
]),
method: 'GET',
headers: ({ username, password }) => buildAuthHeaders(username, password),
stripAuthOnRedirect: true,
},
transformResponse: async (response, params) => {
const exchangeName = params?.exchange ?? ''
if (!response.ok) {
const error = await extractErrorMessage(response)
return { success: false, output: { exchangeName, bindings: [], count: 0 }, error }
}
const data = await response.json()
const bindings = Array.isArray(data) ? data.map(projectBinding) : []
return { success: true, output: { exchangeName, bindings, count: bindings.length } }
},
outputs: {
exchangeName: { type: 'string', description: 'Exchange the bindings originate from' },
bindings: {
type: 'array',
description:
'Bindings routing out of the exchange. An empty list means nothing it publishes can be delivered',
items: { type: 'object', properties: RABBITMQ_BINDING_OUTPUT_PROPERTIES },
},
count: { type: 'number', description: 'Number of bindings returned' },
},
}
+110
View File
@@ -0,0 +1,110 @@
import type {
RabbitmqExchange,
RabbitmqListExchangesParams,
RabbitmqListExchangesResponse,
} from '@/tools/rabbitmq/types'
import {
buildAuthHeaders,
buildManagementUrl,
clampPageSize,
extractErrorMessage,
projectExchange,
RABBITMQ_CONNECTION_PARAMS,
RABBITMQ_EXCHANGE_COLUMNS,
RABBITMQ_EXCHANGE_OUTPUT_PROPERTIES,
RABBITMQ_MAX_PAGE_SIZE,
resolveVhost,
unwrapPaginated,
} from '@/tools/rabbitmq/utils'
import type { ToolConfig } from '@/tools/types'
const DEFAULT_PAGE_SIZE = 50
export const rabbitmqListExchangesTool: ToolConfig<
RabbitmqListExchangesParams,
RabbitmqListExchangesResponse
> = {
id: 'rabbitmq_list_exchanges',
name: 'RabbitMQ List Exchanges',
description:
'List exchanges in a RabbitMQ virtual host with their type and declaration settings.',
version: '1.0.0',
params: {
...RABBITMQ_CONNECTION_PARAMS,
page: {
type: 'number',
required: false,
visibility: 'user-or-llm',
description: 'Page of results to return, starting at 1',
},
pageSize: {
type: 'number',
required: false,
visibility: 'user-or-llm',
description: `Exchanges per page, from 1 to ${RABBITMQ_MAX_PAGE_SIZE}. Defaults to ${DEFAULT_PAGE_SIZE}`,
},
name: {
type: 'string',
required: false,
visibility: 'user-or-llm',
description: 'Filter exchanges whose name contains this value',
},
useRegex: {
type: 'boolean',
required: false,
visibility: 'user-only',
description: 'Treat the name filter as a regular expression',
},
},
request: {
url: ({ host, vhost, name, page, pageSize, useRegex }) =>
buildManagementUrl(host, ['exchanges', resolveVhost(vhost)], {
page: page ?? 1,
page_size: clampPageSize(pageSize, DEFAULT_PAGE_SIZE),
name: name,
use_regex: useRegex ? 'true' : undefined,
columns: RABBITMQ_EXCHANGE_COLUMNS,
}),
method: 'GET',
headers: ({ username, password }) => buildAuthHeaders(username, password),
stripAuthOnRedirect: true,
},
transformResponse: async (response) => {
if (!response.ok) {
const error = await extractErrorMessage(response)
return {
success: false,
output: { exchanges: [], count: 0, totalCount: null, page: null, pageCount: null },
error,
}
}
const data = await response.json()
const { items, totalCount, page, pageCount } = unwrapPaginated<RabbitmqExchange>(data)
const exchanges = items.map(projectExchange)
return {
success: true,
output: { exchanges, count: exchanges.length, totalCount, page, pageCount },
}
},
outputs: {
exchanges: {
type: 'array',
description: 'Exchanges in the virtual host',
items: { type: 'object', properties: RABBITMQ_EXCHANGE_OUTPUT_PROPERTIES },
},
count: { type: 'number', description: 'Number of exchanges returned on this page' },
totalCount: {
type: 'number',
description: 'Total exchanges in the virtual host before filtering',
optional: true,
},
page: { type: 'number', description: 'Page number returned', optional: true },
pageCount: { type: 'number', description: 'Total number of pages', optional: true },
},
}
+50
View File
@@ -0,0 +1,50 @@
import type { RabbitmqListNodesParams, RabbitmqListNodesResponse } from '@/tools/rabbitmq/types'
import {
buildAuthHeaders,
buildManagementUrl,
extractErrorMessage,
projectNode,
RABBITMQ_CONNECTION_PARAMS,
RABBITMQ_NODE_COLUMNS,
RABBITMQ_NODE_OUTPUT_PROPERTIES,
} from '@/tools/rabbitmq/utils'
import type { ToolConfig } from '@/tools/types'
export const rabbitmqListNodesTool: ToolConfig<RabbitmqListNodesParams, RabbitmqListNodesResponse> =
{
id: 'rabbitmq_list_nodes',
name: 'RabbitMQ List Nodes',
description:
'List the cluster nodes with memory, disk, file-descriptor, and alarm state. A fired alarm blocks publishers broker-wide.',
version: '1.0.0',
params: { ...RABBITMQ_CONNECTION_PARAMS },
request: {
url: ({ host }) => buildManagementUrl(host, ['nodes'], { columns: RABBITMQ_NODE_COLUMNS }),
method: 'GET',
headers: ({ username, password }) => buildAuthHeaders(username, password),
stripAuthOnRedirect: true,
},
transformResponse: async (response) => {
if (!response.ok) {
const error = await extractErrorMessage(response)
return { success: false, output: { nodes: [], count: 0 }, error }
}
const data = await response.json()
const nodes = Array.isArray(data) ? data.map(projectNode) : []
return { success: true, output: { nodes, count: nodes.length } }
},
outputs: {
nodes: {
type: 'array',
description: 'Cluster nodes and their resource headroom',
items: { type: 'object', properties: RABBITMQ_NODE_OUTPUT_PROPERTIES },
},
count: { type: 'number', description: 'Number of nodes in the cluster' },
},
}
+55
View File
@@ -0,0 +1,55 @@
import type {
RabbitmqListPoliciesParams,
RabbitmqListPoliciesResponse,
} from '@/tools/rabbitmq/types'
import {
buildAuthHeaders,
buildManagementUrl,
extractErrorMessage,
projectPolicy,
RABBITMQ_CONNECTION_PARAMS,
RABBITMQ_POLICY_OUTPUT_PROPERTIES,
resolveVhost,
} from '@/tools/rabbitmq/utils'
import type { ToolConfig } from '@/tools/types'
export const rabbitmqListPoliciesTool: ToolConfig<
RabbitmqListPoliciesParams,
RabbitmqListPoliciesResponse
> = {
id: 'rabbitmq_list_policies',
name: 'RabbitMQ List Policies',
description:
'List the policies in a virtual host. Policies are how dead-lettering, TTLs, and length limits get applied to matching queues and exchanges.',
version: '1.0.0',
params: { ...RABBITMQ_CONNECTION_PARAMS },
request: {
url: ({ host, vhost }) => buildManagementUrl(host, ['policies', resolveVhost(vhost)]),
method: 'GET',
headers: ({ username, password }) => buildAuthHeaders(username, password),
stripAuthOnRedirect: true,
},
transformResponse: async (response) => {
if (!response.ok) {
const error = await extractErrorMessage(response)
return { success: false, output: { policies: [], count: 0 }, error }
}
const data = await response.json()
const policies = Array.isArray(data) ? data.map(projectPolicy) : []
return { success: true, output: { policies, count: policies.length } }
},
outputs: {
policies: {
type: 'array',
description: 'Policies defined in the virtual host',
items: { type: 'object', properties: RABBITMQ_POLICY_OUTPUT_PROPERTIES },
},
count: { type: 'number', description: 'Number of policies returned' },
},
}
+110
View File
@@ -0,0 +1,110 @@
import type {
RabbitmqListQueuesParams,
RabbitmqListQueuesResponse,
RabbitmqQueue,
} from '@/tools/rabbitmq/types'
import {
buildAuthHeaders,
buildManagementUrl,
clampPageSize,
extractErrorMessage,
projectQueue,
RABBITMQ_CONNECTION_PARAMS,
RABBITMQ_MAX_PAGE_SIZE,
RABBITMQ_QUEUE_COLUMNS,
RABBITMQ_QUEUE_OUTPUT_PROPERTIES,
resolveVhost,
unwrapPaginated,
} from '@/tools/rabbitmq/utils'
import type { ToolConfig } from '@/tools/types'
const DEFAULT_PAGE_SIZE = 50
export const rabbitmqListQueuesTool: ToolConfig<
RabbitmqListQueuesParams,
RabbitmqListQueuesResponse
> = {
id: 'rabbitmq_list_queues',
name: 'RabbitMQ List Queues',
description:
'List queues in a RabbitMQ virtual host with their depth, consumer count, and configuration.',
version: '1.0.0',
params: {
...RABBITMQ_CONNECTION_PARAMS,
page: {
type: 'number',
required: false,
visibility: 'user-or-llm',
description: 'Page of results to return, starting at 1',
},
pageSize: {
type: 'number',
required: false,
visibility: 'user-or-llm',
description: `Queues per page, from 1 to ${RABBITMQ_MAX_PAGE_SIZE}. Defaults to ${DEFAULT_PAGE_SIZE}`,
},
name: {
type: 'string',
required: false,
visibility: 'user-or-llm',
description: 'Filter queues whose name contains this value',
},
useRegex: {
type: 'boolean',
required: false,
visibility: 'user-only',
description: 'Treat the name filter as a regular expression',
},
},
request: {
url: ({ host, vhost, name, page, pageSize, useRegex }) =>
buildManagementUrl(host, ['queues', resolveVhost(vhost)], {
page: page ?? 1,
page_size: clampPageSize(pageSize, DEFAULT_PAGE_SIZE),
name: name,
use_regex: useRegex ? 'true' : undefined,
columns: RABBITMQ_QUEUE_COLUMNS,
}),
method: 'GET',
headers: ({ username, password }) => buildAuthHeaders(username, password),
stripAuthOnRedirect: true,
},
transformResponse: async (response) => {
if (!response.ok) {
const error = await extractErrorMessage(response)
return {
success: false,
output: { queues: [], count: 0, totalCount: null, page: null, pageCount: null },
error,
}
}
const data = await response.json()
const { items, totalCount, page, pageCount } = unwrapPaginated<RabbitmqQueue>(data)
const queues = items.map(projectQueue)
return {
success: true,
output: { queues, count: queues.length, totalCount, page, pageCount },
}
},
outputs: {
queues: {
type: 'array',
description: 'Queues in the virtual host',
items: { type: 'object', properties: RABBITMQ_QUEUE_OUTPUT_PROPERTIES },
},
count: { type: 'number', description: 'Number of queues returned on this page' },
totalCount: {
type: 'number',
description: 'Total queues in the virtual host before filtering',
optional: true,
},
page: { type: 'number', description: 'Page number returned', optional: true },
pageCount: { type: 'number', description: 'Total number of pages', optional: true },
},
}
+55
View File
@@ -0,0 +1,55 @@
import type { RabbitmqListVhostsParams, RabbitmqListVhostsResponse } from '@/tools/rabbitmq/types'
import {
buildAuthHeaders,
buildManagementUrl,
extractErrorMessage,
projectVhost,
RABBITMQ_CONNECTION_PARAMS,
RABBITMQ_VHOST_OUTPUT_PROPERTIES,
} from '@/tools/rabbitmq/utils'
import type { ToolConfig } from '@/tools/types'
export const rabbitmqListVhostsTool: ToolConfig<
RabbitmqListVhostsParams,
RabbitmqListVhostsResponse
> = {
id: 'rabbitmq_list_vhosts',
name: 'RabbitMQ List Virtual Hosts',
description:
'List the virtual hosts on the broker with their message totals, so you can discover which scopes exist.',
version: '1.0.0',
params: { ...RABBITMQ_CONNECTION_PARAMS },
request: {
/**
* Deliberately sent without pagination parameters: unlike the other list endpoints, this one
* responds with HTTP 500 when `page`/`page_size` are supplied.
*/
url: ({ host }) => buildManagementUrl(host, ['vhosts']),
method: 'GET',
headers: ({ username, password }) => buildAuthHeaders(username, password),
stripAuthOnRedirect: true,
},
transformResponse: async (response) => {
if (!response.ok) {
const error = await extractErrorMessage(response)
return { success: false, output: { vhosts: [], count: 0 }, error }
}
const data = await response.json()
const vhosts = Array.isArray(data) ? data.map(projectVhost) : []
return { success: true, output: { vhosts, count: vhosts.length } }
},
outputs: {
vhosts: {
type: 'array',
description: 'Virtual hosts the authenticated user can see',
items: { type: 'object', properties: RABBITMQ_VHOST_OUTPUT_PROPERTIES },
},
count: { type: 'number', description: 'Number of virtual hosts returned' },
},
}
+138
View File
@@ -0,0 +1,138 @@
import type {
RabbitmqPublishMessageParams,
RabbitmqPublishMessageResponse,
} from '@/tools/rabbitmq/types'
import {
buildAuthHeaders,
buildManagementUrl,
extractErrorMessage,
parseJsonObjectParam,
RABBITMQ_CONNECTION_PARAMS,
resolveVhost,
} from '@/tools/rabbitmq/utils'
import type { ToolConfig } from '@/tools/types'
export const rabbitmqPublishMessageTool: ToolConfig<
RabbitmqPublishMessageParams,
RabbitmqPublishMessageResponse
> = {
id: 'rabbitmq_publish_message',
name: 'RabbitMQ Publish Message',
description:
'Publish a message to a RabbitMQ exchange with a routing key. Reports whether the message was routed to at least one queue.',
version: '1.0.0',
params: {
...RABBITMQ_CONNECTION_PARAMS,
exchange: {
type: 'string',
required: false,
visibility: 'user-or-llm',
description:
'Exchange to publish to. Leave empty to publish to the default exchange, which routes by queue name. Empty is a valid value, so this is not required.',
},
routingKey: {
type: 'string',
required: true,
visibility: 'user-or-llm',
description:
'Routing key. When publishing to the default exchange this is the target queue name.',
},
payload: {
type: 'string',
required: true,
visibility: 'user-or-llm',
description: 'Message body to publish',
},
payloadEncoding: {
type: 'string',
required: false,
visibility: 'user-only',
description: 'How the payload is encoded: string (default) or base64',
},
properties: {
type: 'string',
required: false,
visibility: 'user-or-llm',
description:
'AMQP basic properties as a JSON object, e.g. {"delivery_mode":2,"content_type":"application/json"}',
},
headers: {
type: 'string',
required: false,
visibility: 'user-or-llm',
description: 'Message headers as a JSON object, e.g. {"source":"sim"}',
},
},
request: {
url: ({ host, vhost, exchange }) =>
buildManagementUrl(host, ['exchanges', resolveVhost(vhost), exchange ?? '', 'publish']),
method: 'POST',
headers: ({ username, password }) => buildAuthHeaders(username, password),
stripAuthOnRedirect: true,
body: ({
headers: rawHeaders,
payload,
payloadEncoding,
properties: rawProperties,
routingKey,
}) => {
const properties = parseJsonObjectParam(rawProperties, 'properties') ?? {}
const headers = parseJsonObjectParam(rawHeaders, 'headers')
if (headers) {
// Only merge onto an existing headers object. Spreading a string or array here would
// turn it into index-keyed junk on the published message.
const existing = properties.headers
const base =
typeof existing === 'object' && existing !== null && !Array.isArray(existing)
? (existing as Record<string, unknown>)
: {}
properties.headers = { ...base, ...headers }
}
return {
properties,
routing_key: routingKey,
payload: payload,
payload_encoding: payloadEncoding === 'base64' ? 'base64' : 'string',
}
},
},
transformResponse: async (response, params) => {
if (!response.ok) {
const error = await extractErrorMessage(response)
return {
success: false,
output: {
routed: false,
exchange: params?.exchange ?? '',
routingKey: params?.routingKey ?? '',
},
error,
}
}
const data = await response.json()
return {
success: true,
output: {
routed: data?.routed === true,
exchange: params?.exchange ?? '',
routingKey: params?.routingKey ?? '',
},
}
},
outputs: {
routed: {
type: 'boolean',
description:
'Whether the message was routed to at least one queue. False means no binding matched and the message was dropped.',
},
exchange: { type: 'string', description: 'Exchange the message was published to' },
routingKey: { type: 'string', description: 'Routing key the message was published with' },
},
}
+56
View File
@@ -0,0 +1,56 @@
import type { RabbitmqPurgeQueueParams, RabbitmqPurgeQueueResponse } from '@/tools/rabbitmq/types'
import {
buildAuthHeaders,
buildManagementUrl,
extractErrorMessage,
RABBITMQ_CONNECTION_PARAMS,
resolveVhost,
} from '@/tools/rabbitmq/utils'
import type { ToolConfig } from '@/tools/types'
export const rabbitmqPurgeQueueTool: ToolConfig<
RabbitmqPurgeQueueParams,
RabbitmqPurgeQueueResponse
> = {
id: 'rabbitmq_purge_queue',
name: 'RabbitMQ Purge Queue',
description:
'Discard every ready message in a RabbitMQ queue while leaving the queue itself in place.',
version: '1.0.0',
params: {
...RABBITMQ_CONNECTION_PARAMS,
queue: {
type: 'string',
required: true,
visibility: 'user-or-llm',
description: 'Name of the queue to purge',
},
},
request: {
url: ({ host, vhost, queue }) =>
buildManagementUrl(host, ['queues', resolveVhost(vhost), queue, 'contents']),
method: 'DELETE',
headers: ({ username, password }) => buildAuthHeaders(username, password),
stripAuthOnRedirect: true,
},
transformResponse: async (response, params) => {
const queueName = params?.queue ?? ''
const vhost = params ? resolveVhost(params.vhost) : ''
if (!response.ok) {
const error = await extractErrorMessage(response)
return { success: false, output: { queueName, vhost, purged: false }, error }
}
return { success: true, output: { queueName, vhost, purged: true } }
},
outputs: {
queueName: { type: 'string', description: 'Name of the purged queue' },
vhost: { type: 'string', description: 'Virtual host the queue belongs to' },
purged: { type: 'boolean', description: 'Whether the queue was purged' },
},
}
+454
View File
@@ -0,0 +1,454 @@
/**
* @vitest-environment node
*/
import { describe, expect, it } from 'vitest'
import * as rabbitmqTools from '@/tools/rabbitmq'
import { rabbitmqCreateBindingTool } from '@/tools/rabbitmq/create_binding'
import { rabbitmqCreatePolicyTool } from '@/tools/rabbitmq/create_policy'
import { rabbitmqDeleteBindingTool } from '@/tools/rabbitmq/delete_binding'
import { rabbitmqDeleteQueueTool } from '@/tools/rabbitmq/delete_queue'
import { rabbitmqGetMessagesTool } from '@/tools/rabbitmq/get_messages'
import { rabbitmqGetQueueTool } from '@/tools/rabbitmq/get_queue'
import { rabbitmqHealthCheckTool } from '@/tools/rabbitmq/health_check'
import { rabbitmqListQueuesTool } from '@/tools/rabbitmq/list_queues'
import { rabbitmqListVhostsTool } from '@/tools/rabbitmq/list_vhosts'
import { rabbitmqPublishMessageTool } from '@/tools/rabbitmq/publish_message'
const conn = {
host: 'https://rabbit.example.com:15672',
username: 'guest',
password: 'guest',
}
function url(tool: { request: { url: unknown } }, params: Record<string, unknown>): string {
const build = tool.request.url as (p: Record<string, unknown>) => string
return build({ ...conn, ...params })
}
describe('rabbitmq management url building', () => {
it('encodes the default vhost as %2F', () => {
expect(url(rabbitmqGetQueueTool, { queue: 'orders' })).toBe(
'https://rabbit.example.com:15672/api/queues/%2F/orders'
)
})
it('encodes a named vhost and a queue name containing a slash', () => {
expect(url(rabbitmqGetQueueTool, { vhost: 'prod/eu', queue: 'orders/new' })).toBe(
'https://rabbit.example.com:15672/api/queues/prod%2Feu/orders%2Fnew'
)
})
it('tolerates a host copied with a trailing slash or /api suffix', () => {
const expected = 'https://rabbit.example.com:15672/api/queues/%2F/orders'
expect(url(rabbitmqGetQueueTool, { host: `${conn.host}/`, queue: 'orders' })).toBe(expected)
expect(url(rabbitmqGetQueueTool, { host: `${conn.host}/api`, queue: 'orders' })).toBe(expected)
expect(url(rabbitmqGetQueueTool, { host: `${conn.host}/api/`, queue: 'orders' })).toBe(expected)
})
it('rejects a host that is not an http(s) URL', () => {
expect(() => url(rabbitmqGetQueueTool, { host: 'rabbit.example.com', queue: 'q' })).toThrow(
/Provide a full URL/
)
expect(() =>
url(rabbitmqGetQueueTool, { host: 'amqp://rabbit.example.com', queue: 'q' })
).toThrow(/Unsupported RabbitMQ host protocol/)
})
it('omits delete guards that were not requested', () => {
expect(url(rabbitmqDeleteQueueTool, { queue: 'orders' })).toBe(
'https://rabbit.example.com:15672/api/queues/%2F/orders'
)
expect(url(rabbitmqDeleteQueueTool, { queue: 'orders', ifEmpty: true })).toBe(
'https://rabbit.example.com:15672/api/queues/%2F/orders?if-empty=true'
)
})
})
describe('rabbitmq_publish_message', () => {
const body = (params: Record<string, unknown>) =>
rabbitmqPublishMessageTool.request.body?.({ ...conn, ...params } as never) as Record<
string,
unknown
>
it('merges headers into the AMQP properties', () => {
expect(
body({
exchange: 'orders',
routingKey: 'created',
payload: 'body',
properties: '{"delivery_mode":2,"headers":{"a":"1"}}',
headers: '{"b":"2"}',
})
).toEqual({
properties: { delivery_mode: 2, headers: { a: '1', b: '2' } },
routing_key: 'created',
payload: 'body',
payload_encoding: 'string',
})
})
it('defaults to string encoding and empty properties', () => {
expect(body({ exchange: '', routingKey: 'orders', payload: 'hi' })).toEqual({
properties: {},
routing_key: 'orders',
payload: 'hi',
payload_encoding: 'string',
})
})
it('rejects properties that are not a JSON object', () => {
expect(() =>
body({ exchange: '', routingKey: 'q', payload: 'hi', properties: '[1,2]' })
).toThrow(/properties must be a JSON object/)
expect(() => body({ exchange: '', routingKey: 'q', payload: 'hi', headers: 'nope' })).toThrow(
/headers must be a valid JSON object/
)
})
})
describe('rabbitmq response handling', () => {
it('reads the pagination envelope returned when page params are sent', async () => {
const response = new Response(
JSON.stringify({
items: [{ name: 'orders', vhost: '/', messages: 3 }],
total_count: 12,
page: 2,
page_count: 4,
}),
{ status: 200 }
)
const result = await rabbitmqListQueuesTool.transformResponse!(response)
expect(result.success).toBe(true)
expect(result.output.totalCount).toBe(12)
expect(result.output.page).toBe(2)
expect(result.output.pageCount).toBe(4)
expect(result.output.queues[0]).toMatchObject({ name: 'orders', messages: 3 })
})
it('reads a bare array from a broker that returned no envelope', async () => {
const response = new Response(JSON.stringify([{ name: 'orders', vhost: '/' }]), { status: 200 })
const result = await rabbitmqListQueuesTool.transformResponse!(response)
expect(result.output.count).toBe(1)
expect(result.output.page).toBeNull()
})
it('nulls queue statistics the broker has not collected yet', async () => {
const response = new Response(
JSON.stringify({ name: 'fresh', vhost: '/', durable: true, type: 'classic' }),
{ status: 200 }
)
const result = await rabbitmqGetQueueTool.transformResponse!(response)
expect(result.output.queue).toMatchObject({
name: 'fresh',
durable: true,
messages: null,
consumers: null,
})
})
it('surfaces the broker error and reason on a failed request', async () => {
const response = new Response(JSON.stringify({ error: 'not_found', reason: 'no queue' }), {
status: 404,
})
const result = await rabbitmqGetQueueTool.transformResponse!(response)
expect(result.success).toBe(false)
expect(result.error).toBe('RabbitMQ request failed (404): not_found: no queue')
})
it('decodes the binding properties key out of the location header', async () => {
const response = new Response(null, {
status: 201,
headers: { location: 'orders/sim.a%252Fb' },
})
const result = await rabbitmqCreateBindingTool.transformResponse!(response, {
...conn,
exchange: 'amq.topic',
queue: 'orders',
routingKey: 'sim.a/b',
})
expect(result.output.queueName).toBe('orders')
expect(result.output.propertiesKey).toBe('sim.a%2Fb')
expect(result.output.created).toBe(true)
})
it('returns a null properties key when the broker sent no location header', async () => {
const response = new Response(null, { status: 201 })
const result = await rabbitmqCreateBindingTool.transformResponse!(response, {
...conn,
exchange: 'amq.topic',
queue: 'orders',
})
expect(result.output.propertiesKey).toBeNull()
})
})
describe('rabbitmq_get_messages bounds', () => {
const body = (params: Record<string, unknown>) =>
rabbitmqGetMessagesTool.request.body?.({ ...conn, ...params } as never) as Record<
string,
unknown
>
it('defaults to one message and a byte-capped payload', () => {
expect(body({ queue: 'orders' })).toEqual({
count: 1,
ackmode: 'ack_requeue_true',
encoding: 'auto',
truncate: 50_000,
})
})
it('caps a count the broker would otherwise accept unbounded', () => {
expect(body({ queue: 'orders', count: 100_000 }).count).toBe(50)
expect(body({ queue: 'orders', count: 0 }).count).toBe(1)
expect(body({ queue: 'orders', count: 25 }).count).toBe(25)
})
it('falls back to the safe acknowledgement mode for an unknown value', () => {
expect(body({ queue: 'orders', ackmode: 'destroy_everything' }).ackmode).toBe(
'ack_requeue_true'
)
expect(body({ queue: 'orders', ackmode: 'ack_requeue_false' }).ackmode).toBe(
'ack_requeue_false'
)
})
it('flags a payload the broker cut short at the truncate limit', async () => {
const response = new Response(
JSON.stringify([
{ payload: 'X'.repeat(100), payload_bytes: 5000, payload_encoding: 'string' },
{ payload: 'small', payload_bytes: 5, payload_encoding: 'string' },
]),
{ status: 200 }
)
const result = await rabbitmqGetMessagesTool.transformResponse!(response, {
...conn,
queue: 'orders',
truncate: 100,
})
expect(result.output.messages[0]).toMatchObject({ payloadBytes: 5000, truncated: true })
expect(result.output.messages[1]).toMatchObject({ payloadBytes: 5, truncated: false })
})
})
describe('rabbitmq list bounds', () => {
it('clamps a page size outside the range the broker accepts', () => {
expect(url(rabbitmqListQueuesTool, { pageSize: 5000 })).toContain('page_size=500')
expect(url(rabbitmqListQueuesTool, { pageSize: 0 })).toContain('page_size=1')
expect(url(rabbitmqListQueuesTool, {})).toContain('page_size=50')
})
it('requests only the columns it projects', () => {
const built = url(rabbitmqListQueuesTool, {})
expect(built).toContain('columns=')
expect(decodeURIComponent(built)).toContain('columns=name,vhost,type,state,durable')
})
})
describe('rabbitmq path segments', () => {
it('trims whitespace pasted around a queue name', () => {
expect(url(rabbitmqGetQueueTool, { queue: ' orders ' })).toBe(
'https://rabbit.example.com:15672/api/queues/%2F/orders'
)
})
})
describe('rabbitmq_health_check', () => {
it('routes each check to its own path, including the ones taking arguments', () => {
const base = 'https://rabbit.example.com:15672/api/health/checks'
expect(url(rabbitmqHealthCheckTool, {})).toBe(`${base}/alarms`)
expect(url(rabbitmqHealthCheckTool, { check: 'virtual-hosts' })).toBe(`${base}/virtual-hosts`)
expect(url(rabbitmqHealthCheckTool, { check: 'port-listener', port: 5672 })).toBe(
`${base}/port-listener/5672`
)
expect(url(rabbitmqHealthCheckTool, { check: 'protocol-listener', protocol: 'amqp/ssl' })).toBe(
`${base}/protocol-listener/amqp%2Fssl`
)
expect(
url(rabbitmqHealthCheckTool, { check: 'certificate-expiration', within: 3, unit: 'weeks' })
).toBe(`${base}/certificate-expiration/3/weeks`)
})
it('names the missing argument instead of building a wrong URL', () => {
expect(() => url(rabbitmqHealthCheckTool, { check: 'port-listener' })).toThrow(
/port is required/
)
expect(() => url(rabbitmqHealthCheckTool, { check: 'protocol-listener' })).toThrow(
/protocol is required/
)
expect(() => url(rabbitmqHealthCheckTool, { check: 'certificate-expiration' })).toThrow(
/within is required/
)
})
it('treats a failing check as a successful call reporting unhealthy', async () => {
const response = new Response(
JSON.stringify({ status: 'failed', reason: 'No active listener', missing: 9999 }),
{ status: 503 }
)
const result = await rabbitmqHealthCheckTool.transformResponse!(response, {
...conn,
check: 'port-listener',
port: 9999,
})
expect(result.success).toBe(true)
expect(result.output.healthy).toBe(false)
expect(result.output.status).toBe('failed')
expect(result.output.reason).toBe('No active listener')
expect(result.output.details).toMatchObject({ missing: 9999 })
})
it('still reports a real transport failure as an error', async () => {
const response = new Response(JSON.stringify({ error: 'not_authorized' }), { status: 401 })
const result = await rabbitmqHealthCheckTool.transformResponse!(response, { ...conn })
expect(result.success).toBe(false)
expect(result.output.healthy).toBe(false)
})
})
describe('rabbitmq binding destinations', () => {
it('addresses a queue destination by default and an exchange when asked', () => {
expect(
url(rabbitmqDeleteBindingTool, {
exchange: 'orders',
destination: 'orders-q',
propertiesKey: 'a.b',
})
).toBe('https://rabbit.example.com:15672/api/bindings/%2F/e/orders/q/orders-q/a.b')
expect(
url(rabbitmqDeleteBindingTool, {
exchange: 'orders',
destination: 'audit-x',
destinationType: 'exchange',
propertiesKey: '~',
})
).toBe('https://rabbit.example.com:15672/api/bindings/%2F/e/orders/e/audit-x/~')
})
})
describe('rabbitmq_list_vhosts', () => {
it('never sends pagination parameters, which this endpoint answers with a 500', () => {
expect(url(rabbitmqListVhostsTool, {})).toBe('https://rabbit.example.com:15672/api/vhosts')
})
})
describe('rabbitmq_create_policy', () => {
const body = (params: Record<string, unknown>) =>
rabbitmqCreatePolicyTool.request.body?.({ ...conn, ...params } as never) as Record<
string,
unknown
>
it('sends the broker its hyphenated apply-to key with a safe default', () => {
expect(
body({ policyName: 'p', pattern: '^orders\\.', definition: '{"message-ttl":1000}' })
).toEqual({
pattern: '^orders\\.',
definition: { 'message-ttl': 1000 },
priority: 0,
'apply-to': 'queues',
})
})
it('rejects an apply-to value the broker does not recognise', () => {
expect(
body({ policyName: 'p', pattern: '.*', definition: '{}', applyTo: 'bogus' })['apply-to']
).toBe('queues')
expect(
body({ policyName: 'p', pattern: '.*', definition: '{}', applyTo: 'quorum_queues' })[
'apply-to'
]
).toBe('quorum_queues')
})
})
describe('rabbitmq credential exposure', () => {
it('strips the Basic auth header on every redirect, for every tool', () => {
const tools = Object.values(rabbitmqTools)
expect(tools.length).toBeGreaterThan(0)
for (const tool of tools) {
expect(`${tool.id}:${tool.request.stripAuthOnRedirect}`).toBe(`${tool.id}:true`)
}
})
it('refuses a remote plain-http host rather than leaving it to fail at dispatch', () => {
expect(() =>
url(rabbitmqGetQueueTool, { host: 'http://rabbit.example.com:15672', queue: 'q' })
).toThrow(/must use https/)
expect(() => url(rabbitmqGetQueueTool, { host: 'http://10.0.0.5:15672', queue: 'q' })).toThrow(
/must use https/
)
})
it('still allows a loopback host over plain http for local brokers', () => {
expect(url(rabbitmqGetQueueTool, { host: 'http://localhost:15672', queue: 'q' })).toBe(
'http://localhost:15672/api/queues/%2F/q'
)
expect(url(rabbitmqGetQueueTool, { host: 'http://127.0.0.1:15672', queue: 'q' })).toBe(
'http://127.0.0.1:15672/api/queues/%2F/q'
)
expect(
url(rabbitmqGetQueueTool, { host: 'https://rabbit.example.com:15672', queue: 'q' })
).toBe('https://rabbit.example.com:15672/api/queues/%2F/q')
})
})
describe('rabbitmq_get_messages response budget', () => {
const body = (params: Record<string, unknown>) =>
rabbitmqGetMessagesTool.request.body?.({ ...conn, ...params } as never) as Record<
string,
number
>
const TRANSPORT_CAP = 10 * 1024 * 1024
// `truncate` bounds the payload only — the broker returns AMQP properties in full — so the
// worst case a batch can produce is every message also carrying a full content header frame.
const FRAME_MAX = 131_072
it('keeps payloads plus worst-case message metadata under the transport cap', () => {
for (const count of [1, 2, 10, 25, 50, 100]) {
const sent = body({ queue: 'q', count, truncate: 100_000_000 })
expect(sent.count * (sent.truncate * (4 / 3) + FRAME_MAX)).toBeLessThan(TRANSPORT_CAP)
}
})
it('caps the batch so reserved metadata alone cannot exhaust the budget', () => {
expect(body({ queue: 'q', count: 100_000 }).count).toBe(50)
expect(body({ queue: 'q', count: 100_000 }).count * FRAME_MAX).toBeLessThan(TRANSPORT_CAP)
})
it('caps a single oversized truncate request', () => {
expect(body({ queue: 'q', truncate: 100_000_000 }).truncate).toBe(1_000_000)
})
it('leaves a reasonable truncate untouched at a low count', () => {
expect(body({ queue: 'q', truncate: 20_000, count: 5 }).truncate).toBe(20_000)
})
it('shortens payloads rather than failing the retrieval when count is high', () => {
expect(body({ queue: 'q', truncate: 50_000, count: 2 }).truncate).toBe(50_000)
const high = body({ queue: 'q', truncate: 50_000, count: 50 }).truncate
expect(high).toBeLessThan(50_000)
expect(high).toBeGreaterThanOrEqual(1_024)
})
})
+547
View File
@@ -0,0 +1,547 @@
import type { ToolResponse } from '@/tools/types'
/**
* Connection parameters shared by every RabbitMQ tool. All operations go through the
* RabbitMQ Management HTTP API, which authenticates with HTTP basic auth.
*
* @see https://www.rabbitmq.com/docs/management#http-api
*/
interface RabbitmqBaseParams {
/** Management API base URL, e.g. `https://rabbit.example.com:15672`. */
host: string
username: string
password: string
/** Virtual host the operation targets. Defaults to `/`. */
vhost?: string
}
export interface RabbitmqPublishMessageParams extends RabbitmqBaseParams {
/** Empty targets the default exchange, which routes by queue name. */
exchange?: string
routingKey: string
payload: string
payloadEncoding?: 'string' | 'base64'
/** AMQP basic properties as a JSON object string, e.g. `{"delivery_mode":2}`. */
properties?: string
/** Message headers as a JSON object string. Merged into the AMQP properties. */
headers?: string
}
export interface RabbitmqGetMessagesParams extends RabbitmqBaseParams {
queue: string
count?: number
ackmode?:
| 'ack_requeue_true'
| 'ack_requeue_false'
| 'reject_requeue_true'
| 'reject_requeue_false'
encoding?: 'auto' | 'base64'
/** Truncate payloads longer than this many bytes. */
truncate?: number
}
export interface RabbitmqListQueuesParams extends RabbitmqBaseParams {
page?: number
pageSize?: number
/** Filters returned queues by name. Treated as a regular expression when `useRegex` is set. */
name?: string
useRegex?: boolean
}
export interface RabbitmqGetQueueParams extends RabbitmqBaseParams {
queue: string
}
export interface RabbitmqCreateQueueParams extends RabbitmqBaseParams {
queue: string
durable?: boolean
autoDelete?: boolean
/** Optional queue arguments as a JSON object string, e.g. `{"x-queue-type":"quorum"}`. */
arguments?: string
}
export interface RabbitmqDeleteQueueParams extends RabbitmqBaseParams {
queue: string
ifUnused?: boolean
ifEmpty?: boolean
}
export interface RabbitmqPurgeQueueParams extends RabbitmqBaseParams {
queue: string
}
export interface RabbitmqGetExchangeParams extends RabbitmqBaseParams {
exchange?: string
}
export interface RabbitmqCreateExchangeParams extends RabbitmqBaseParams {
exchange: string
exchangeType?: 'direct' | 'fanout' | 'topic' | 'headers'
durable?: boolean
autoDelete?: boolean
internal?: boolean
/** Optional exchange arguments as a JSON object string, e.g. `{"alternate-exchange":"dlx"}`. */
arguments?: string
}
export interface RabbitmqDeleteExchangeParams extends RabbitmqBaseParams {
exchange: string
ifUnused?: boolean
}
export interface RabbitmqListExchangeBindingsParams extends RabbitmqBaseParams {
exchange?: string
}
export interface RabbitmqDeleteBindingParams extends RabbitmqBaseParams {
exchange: string
destination: string
destinationType?: 'queue' | 'exchange'
/** Broker identifier for the binding, as returned by Create Binding or List Bindings. */
propertiesKey: string
}
export type RabbitmqListVhostsParams = RabbitmqBaseParams
export interface RabbitmqListConnectionsParams extends RabbitmqBaseParams {
page?: number
pageSize?: number
name?: string
useRegex?: boolean
}
export interface RabbitmqListChannelsParams extends RabbitmqBaseParams {
page?: number
pageSize?: number
name?: string
useRegex?: boolean
}
export type RabbitmqListConsumersParams = RabbitmqBaseParams
export type RabbitmqListNodesParams = RabbitmqBaseParams
export interface RabbitmqHealthCheckParams extends RabbitmqBaseParams {
check?:
| 'alarms'
| 'local-alarms'
| 'virtual-hosts'
| 'node-is-quorum-critical'
| 'port-listener'
| 'protocol-listener'
| 'certificate-expiration'
port?: number
protocol?: string
within?: number
unit?: 'days' | 'weeks' | 'months' | 'years'
}
export type RabbitmqListPoliciesParams = RabbitmqBaseParams
export interface RabbitmqCreatePolicyParams extends RabbitmqBaseParams {
policyName: string
pattern: string
/** Policy definition as a JSON object string, e.g. `{"dead-letter-exchange":"dlx"}`. */
definition: string
priority?: number
applyTo?: 'all' | 'queues' | 'classic_queues' | 'quorum_queues' | 'streams' | 'exchanges'
}
export interface RabbitmqDeletePolicyParams extends RabbitmqBaseParams {
policyName: string
}
export interface RabbitmqListExchangesParams extends RabbitmqBaseParams {
page?: number
pageSize?: number
name?: string
useRegex?: boolean
}
export interface RabbitmqListBindingsParams extends RabbitmqBaseParams {
queue: string
}
export interface RabbitmqCreateBindingParams extends RabbitmqBaseParams {
exchange: string
queue: string
destinationType?: 'queue' | 'exchange'
routingKey?: string
/** Optional binding arguments as a JSON object string. */
arguments?: string
}
export type RabbitmqGetOverviewParams = RabbitmqBaseParams
/** Projected queue record. Statistics fields are absent until the broker collects stats. */
export interface RabbitmqQueue {
name: string
vhost: string
type: string | null
state: string | null
durable: boolean | null
autoDelete: boolean | null
exclusive: boolean | null
node: string | null
policy: string | null
arguments: Record<string, unknown>
messages: number | null
messagesReady: number | null
messagesUnacknowledged: number | null
consumers: number | null
memory: number | null
}
export interface RabbitmqExchange {
name: string
vhost: string
type: string
durable: boolean
autoDelete: boolean
internal: boolean
arguments: Record<string, unknown>
}
export interface RabbitmqBinding {
source: string
vhost: string
destination: string
destinationType: string
routingKey: string
propertiesKey: string
arguments: Record<string, unknown>
}
export interface RabbitmqMessage {
payload: string
/** True when the broker cut the payload short at the requested truncate limit. */
truncated: boolean
payloadBytes: number
payloadEncoding: string
exchange: string
routingKey: string
redelivered: boolean
/** Messages still left in the queue after this one was retrieved. */
messageCount: number
properties: Record<string, unknown>
}
export interface RabbitmqPublishMessageResponse extends ToolResponse {
output: {
routed: boolean
exchange: string
routingKey: string
}
}
export interface RabbitmqGetMessagesResponse extends ToolResponse {
output: {
queueName: string
count: number
messages: RabbitmqMessage[]
}
}
export interface RabbitmqListQueuesResponse extends ToolResponse {
output: {
queues: RabbitmqQueue[]
count: number
totalCount: number | null
page: number | null
pageCount: number | null
}
}
export interface RabbitmqGetQueueResponse extends ToolResponse {
output: {
queue: RabbitmqQueue
}
}
export interface RabbitmqCreateQueueResponse extends ToolResponse {
output: {
queueName: string
vhost: string
created: boolean
}
}
export interface RabbitmqDeleteQueueResponse extends ToolResponse {
output: {
queueName: string
vhost: string
deleted: boolean
}
}
export interface RabbitmqPurgeQueueResponse extends ToolResponse {
output: {
queueName: string
vhost: string
purged: boolean
}
}
export interface RabbitmqVhost {
name: string
description: string | null
tags: string[]
defaultQueueType: string | null
tracing: boolean
messages: number | null
messagesReady: number | null
messagesUnacknowledged: number | null
clusterState: Record<string, unknown>
}
export interface RabbitmqConnection {
name: string
user: string
vhost: string
state: string | null
protocol: string | null
node: string | null
channels: number | null
peerHost: string | null
peerPort: number | null
connectedAt: number | null
ssl: boolean
}
export interface RabbitmqChannel {
name: string
number: number | null
user: string
vhost: string
node: string | null
state: string | null
consumerCount: number | null
prefetchCount: number | null
messagesUnacknowledged: number | null
confirm: boolean
connectionName: string | null
}
export interface RabbitmqConsumer {
consumerTag: string
queue: string
vhost: string
ackRequired: boolean
active: boolean
activityStatus: string | null
exclusive: boolean
prefetchCount: number | null
channelName: string | null
connectionName: string | null
}
export interface RabbitmqNode {
name: string
type: string | null
running: boolean
memUsed: number | null
memLimit: number | null
memAlarm: boolean
diskFree: number | null
diskFreeLimit: number | null
diskFreeAlarm: boolean
fdUsed: number | null
fdTotal: number | null
procUsed: number | null
procTotal: number | null
uptime: number | null
partitions: string[]
beingDrained: boolean
}
export interface RabbitmqPolicy {
name: string
vhost: string
pattern: string
applyTo: string | null
priority: number | null
definition: Record<string, unknown>
}
export interface RabbitmqGetExchangeResponse extends ToolResponse {
output: {
exchange: RabbitmqExchange
}
}
export interface RabbitmqCreateExchangeResponse extends ToolResponse {
output: {
exchangeName: string
vhost: string
created: boolean
}
}
export interface RabbitmqDeleteExchangeResponse extends ToolResponse {
output: {
exchangeName: string
vhost: string
deleted: boolean
}
}
export interface RabbitmqListExchangeBindingsResponse extends ToolResponse {
output: {
exchangeName: string
bindings: RabbitmqBinding[]
count: number
}
}
export interface RabbitmqDeleteBindingResponse extends ToolResponse {
output: {
exchange: string
destination: string
propertiesKey: string
deleted: boolean
}
}
export interface RabbitmqListVhostsResponse extends ToolResponse {
output: {
vhosts: RabbitmqVhost[]
count: number
}
}
export interface RabbitmqListConnectionsResponse extends ToolResponse {
output: {
connections: RabbitmqConnection[]
count: number
totalCount: number | null
page: number | null
pageCount: number | null
}
}
export interface RabbitmqListChannelsResponse extends ToolResponse {
output: {
channels: RabbitmqChannel[]
count: number
totalCount: number | null
page: number | null
pageCount: number | null
}
}
export interface RabbitmqListConsumersResponse extends ToolResponse {
output: {
consumers: RabbitmqConsumer[]
count: number
}
}
export interface RabbitmqListNodesResponse extends ToolResponse {
output: {
nodes: RabbitmqNode[]
count: number
}
}
export interface RabbitmqHealthCheckResponse extends ToolResponse {
output: {
check: string
healthy: boolean
status: string
reason: string | null
details: Record<string, unknown>
}
}
export interface RabbitmqListPoliciesResponse extends ToolResponse {
output: {
policies: RabbitmqPolicy[]
count: number
}
}
export interface RabbitmqCreatePolicyResponse extends ToolResponse {
output: {
policyName: string
vhost: string
created: boolean
}
}
export interface RabbitmqDeletePolicyResponse extends ToolResponse {
output: {
policyName: string
vhost: string
deleted: boolean
}
}
export interface RabbitmqListExchangesResponse extends ToolResponse {
output: {
exchanges: RabbitmqExchange[]
count: number
totalCount: number | null
page: number | null
pageCount: number | null
}
}
export interface RabbitmqListBindingsResponse extends ToolResponse {
output: {
queueName: string
bindings: RabbitmqBinding[]
count: number
}
}
export interface RabbitmqCreateBindingResponse extends ToolResponse {
output: {
exchange: string
queueName: string
routingKey: string
/** Binding identifier returned by the broker, used to address the binding later. */
propertiesKey: string | null
created: boolean
}
}
export interface RabbitmqGetOverviewResponse extends ToolResponse {
output: {
rabbitmqVersion: string | null
productName: string | null
productVersion: string | null
erlangVersion: string | null
clusterName: string | null
node: string | null
objectTotals: Record<string, unknown>
queueTotals: Record<string, unknown>
messageStats: Record<string, unknown>
}
}
export type RabbitmqResponse =
| RabbitmqPublishMessageResponse
| RabbitmqGetMessagesResponse
| RabbitmqListQueuesResponse
| RabbitmqGetQueueResponse
| RabbitmqCreateQueueResponse
| RabbitmqDeleteQueueResponse
| RabbitmqPurgeQueueResponse
| RabbitmqListExchangesResponse
| RabbitmqListBindingsResponse
| RabbitmqCreateBindingResponse
| RabbitmqGetOverviewResponse
| RabbitmqGetExchangeResponse
| RabbitmqCreateExchangeResponse
| RabbitmqDeleteExchangeResponse
| RabbitmqListExchangeBindingsResponse
| RabbitmqDeleteBindingResponse
| RabbitmqListVhostsResponse
| RabbitmqListConnectionsResponse
| RabbitmqListChannelsResponse
| RabbitmqListConsumersResponse
| RabbitmqListNodesResponse
| RabbitmqHealthCheckResponse
| RabbitmqListPoliciesResponse
| RabbitmqCreatePolicyResponse
| RabbitmqDeletePolicyResponse
+686
View File
@@ -0,0 +1,686 @@
import { isLoopbackIp } from '@sim/security/ssrf'
import type {
RabbitmqBinding,
RabbitmqChannel,
RabbitmqConnection,
RabbitmqConsumer,
RabbitmqExchange,
RabbitmqMessage,
RabbitmqNode,
RabbitmqPolicy,
RabbitmqQueue,
RabbitmqVhost,
} from '@/tools/rabbitmq/types'
/** Connection params every RabbitMQ tool declares. Spread into each tool's `params`. */
export const RABBITMQ_CONNECTION_PARAMS = {
host: {
type: 'string',
required: true,
visibility: 'user-only',
description:
'RabbitMQ Management API base URL, e.g. https://rabbit.example.com:15672. Must use https unless the broker is on a loopback host.',
},
username: {
type: 'string',
required: true,
visibility: 'user-only',
description: 'RabbitMQ username',
},
password: {
type: 'string',
required: true,
visibility: 'user-only',
description: 'RabbitMQ password',
},
vhost: {
type: 'string',
required: false,
visibility: 'user-only',
description: 'Virtual host to operate on. Defaults to /',
},
} as const
/** Output shape of a projected queue. Shared by the list and get tools. */
export const RABBITMQ_QUEUE_OUTPUT_PROPERTIES = {
name: { type: 'string', description: 'Queue name' },
vhost: { type: 'string', description: 'Virtual host the queue belongs to' },
type: { type: 'string', description: 'Queue type, e.g. classic, quorum, or stream' },
state: { type: 'string', description: 'Queue state, e.g. running or idle' },
durable: { type: 'boolean', description: 'Whether the queue survives a broker restart' },
autoDelete: {
type: 'boolean',
description: 'Whether the queue is deleted when its last consumer disconnects',
},
exclusive: { type: 'boolean', description: 'Whether the queue is exclusive to one connection' },
node: { type: 'string', description: 'Cluster node hosting the queue' },
policy: { type: 'string', description: 'Name of the policy applied to the queue, if any' },
arguments: { type: 'json', description: 'Queue arguments the queue was declared with' },
messages: {
type: 'number',
description: 'Total messages in the queue. Null until the broker has collected statistics',
},
messagesReady: {
type: 'number',
description: 'Messages ready for delivery. Null until the broker has collected statistics',
},
messagesUnacknowledged: {
type: 'number',
description: 'Delivered but unacknowledged messages. Null until statistics are collected',
},
consumers: {
type: 'number',
description: 'Number of consumers. Null until the broker has collected statistics',
},
memory: {
type: 'number',
description: 'Memory used by the queue process in bytes. Null until statistics are collected',
},
} as const
export const RABBITMQ_EXCHANGE_OUTPUT_PROPERTIES = {
name: { type: 'string', description: 'Exchange name. Empty string for the default exchange' },
vhost: { type: 'string', description: 'Virtual host the exchange belongs to' },
type: { type: 'string', description: 'Exchange type: direct, fanout, topic, or headers' },
durable: { type: 'boolean', description: 'Whether the exchange survives a broker restart' },
autoDelete: {
type: 'boolean',
description: 'Whether the exchange is deleted when its last binding is removed',
},
internal: {
type: 'boolean',
description: 'Whether the exchange is internal and cannot be published to directly',
},
arguments: { type: 'json', description: 'Exchange arguments the exchange was declared with' },
} as const
export const RABBITMQ_BINDING_OUTPUT_PROPERTIES = {
source: { type: 'string', description: 'Source exchange name' },
vhost: { type: 'string', description: 'Virtual host the binding belongs to' },
destination: { type: 'string', description: 'Destination queue or exchange name' },
destinationType: { type: 'string', description: 'Destination kind: queue or exchange' },
routingKey: { type: 'string', description: 'Routing key the binding matches' },
propertiesKey: { type: 'string', description: 'Broker identifier addressing this binding' },
arguments: { type: 'json', description: 'Binding arguments' },
} as const
export const RABBITMQ_MESSAGE_OUTPUT_PROPERTIES = {
payload: { type: 'string', description: 'Message body' },
payloadBytes: { type: 'number', description: 'Size of the message body in bytes' },
payloadEncoding: { type: 'string', description: 'Payload encoding: string or base64' },
exchange: { type: 'string', description: 'Exchange the message was published to' },
routingKey: { type: 'string', description: 'Routing key the message was published with' },
redelivered: { type: 'boolean', description: 'Whether the message was previously delivered' },
messageCount: {
type: 'number',
description: 'Messages remaining in the queue after this one was retrieved',
},
truncated: {
type: 'boolean',
description:
'Whether the returned payload was cut short by the truncate limit. payloadBytes reports the full size',
},
properties: { type: 'json', description: 'AMQP basic properties, including headers' },
} as const
/**
* Normalizes the user-supplied management host into an origin the API paths append to.
* Tolerates a trailing slash and a trailing `/api` so both forms of the URL people copy
* out of the management UI work.
*/
function normalizeHost(host: string): string {
const trimmed = host?.trim()
if (!trimmed) {
throw new Error('RabbitMQ host is required')
}
let parsed: URL
try {
parsed = new URL(trimmed)
} catch {
throw new Error(
`Invalid RabbitMQ host "${trimmed}". Provide a full URL, e.g. https://rabbit.example.com:15672`
)
}
if (parsed.protocol !== 'http:' && parsed.protocol !== 'https:') {
throw new Error(`Unsupported RabbitMQ host protocol "${parsed.protocol}". Use http or https.`)
}
// Shared outbound validation only permits plain http to a loopback host, so a remote
// http:// endpoint would be refused at dispatch with a generic protocol error. Reject it
// here instead, where the message can name the field the operator has to change.
const hostname = parsed.hostname.toLowerCase().replace(/^\[|\]$/g, '')
const isLoopback = hostname === 'localhost' || isLoopbackIp(hostname)
if (parsed.protocol === 'http:' && !isLoopback) {
throw new Error(
`RabbitMQ host "${trimmed}" must use https. Plain http is only accepted for a loopback host.`
)
}
return `${parsed.origin}${parsed.pathname.replace(/\/(api\/?)?$/, '')}`
}
/** Resolves the target virtual host, defaulting to the broker's default vhost `/`. */
export function resolveVhost(vhost: string | undefined): string {
const trimmed = vhost?.trim()
return trimmed ? trimmed : '/'
}
/**
* The broker rejects a page size outside this range with HTTP 400
* `invalid_pagination_parameters`, so clamp rather than forward an unusable value.
*/
export const RABBITMQ_MAX_PAGE_SIZE = 500
/** Clamps a user- or model-supplied page size into the range the broker accepts. */
export function clampPageSize(pageSize: number | undefined, fallback: number): number {
if (typeof pageSize !== 'number' || !Number.isFinite(pageSize)) return fallback
return Math.min(Math.max(Math.trunc(pageSize), 1), RABBITMQ_MAX_PAGE_SIZE)
}
/**
* Builds a Management API URL. Path segments are encoded individually so vhost `/`
* becomes `%2F`, which the API requires, and trimmed so a name pasted with surrounding
* whitespace does not turn into a 404.
*/
export function buildManagementUrl(
host: string,
segments: string[],
query?: Record<string, string | number | boolean | undefined>
): string {
const path = segments.map((segment) => encodeURIComponent(segment.trim())).join('/')
const url = `${normalizeHost(host)}/api/${path}`
if (!query) return url
const search = new URLSearchParams()
for (const [key, value] of Object.entries(query)) {
if (value !== undefined && value !== '') search.set(key, String(value))
}
const queryString = search.toString()
return queryString ? `${url}?${queryString}` : url
}
export function buildAuthHeaders(username: string, password: string): Record<string, string> {
if (!username || !password) {
throw new Error('RabbitMQ username and password are required')
}
const credentials = Buffer.from(`${username}:${password}`).toString('base64')
return {
Authorization: `Basic ${credentials}`,
'Content-Type': 'application/json',
}
}
/**
* Extracts a readable message from a Management API error body, which is shaped
* `{ "error": "...", "reason": "..." }`.
*/
export async function extractErrorMessage(response: Response): Promise<string> {
const text = await response.text()
let detail = text.trim()
try {
const body = JSON.parse(text)
const reason = typeof body?.reason === 'string' ? body.reason : ''
const error = typeof body?.error === 'string' ? body.error : ''
detail = [error, reason].filter(Boolean).join(': ') || detail
} catch {
// Non-JSON body — fall back to the raw text.
}
return `RabbitMQ request failed (${response.status})${detail ? `: ${detail}` : ''}`
}
/** Parses an optional JSON-object param, surfacing a field-specific message on bad input. */
export function parseJsonObjectParam(
value: string | undefined,
field: string
): Record<string, unknown> | undefined {
const trimmed = value?.trim()
if (!trimmed) return undefined
let parsed: unknown
try {
parsed = JSON.parse(trimmed)
} catch {
throw new Error(`${field} must be a valid JSON object`)
}
if (typeof parsed !== 'object' || parsed === null || Array.isArray(parsed)) {
throw new Error(`${field} must be a JSON object`)
}
return parsed as Record<string, unknown>
}
interface PaginatedResult<T> {
items: T[]
totalCount: number | null
page: number | null
pageCount: number | null
}
/**
* Normalizes a list endpoint response. The Management API returns a pagination envelope
* when page params are supplied and a bare array otherwise.
*/
export function unwrapPaginated<T>(data: unknown): PaginatedResult<T> {
if (Array.isArray(data)) {
return { items: data as T[], totalCount: data.length, page: null, pageCount: null }
}
const envelope = data as Record<string, unknown> | null
const items = Array.isArray(envelope?.items) ? (envelope.items as T[]) : []
return {
items,
totalCount: typeof envelope?.total_count === 'number' ? envelope.total_count : null,
page: typeof envelope?.page === 'number' ? envelope.page : null,
pageCount: typeof envelope?.page_count === 'number' ? envelope.page_count : null,
}
}
function asRecord(value: unknown): Record<string, unknown> {
return typeof value === 'object' && value !== null && !Array.isArray(value)
? (value as Record<string, unknown>)
: {}
}
function asNumberOrNull(value: unknown): number | null {
return typeof value === 'number' ? value : null
}
function asStringOrNull(value: unknown): string | null {
return typeof value === 'string' ? value : null
}
function asBooleanOrNull(value: unknown): boolean | null {
return typeof value === 'boolean' ? value : null
}
/**
* Fields requested from list endpoints via the `columns` query parameter. A full queue record
* carries per-queue statistics, consumer details, and garbage-collection blobs that this
* integration discards, so asking for only the projected fields keeps the response small
* instead of parsing several hundred KB per page and throwing most of it away.
*/
export const RABBITMQ_QUEUE_COLUMNS = [
'name',
'vhost',
'type',
'state',
'durable',
'auto_delete',
'exclusive',
'node',
'policy',
'arguments',
'messages',
'messages_ready',
'messages_unacknowledged',
'consumers',
'memory',
].join(',')
export const RABBITMQ_EXCHANGE_COLUMNS = [
'name',
'vhost',
'type',
'durable',
'auto_delete',
'internal',
'arguments',
].join(',')
/**
* Projects a raw queue object. Message and consumer statistics are omitted by the broker
* until it has collected stats for the queue, so they are nullable.
*/
export function projectQueue(raw: unknown): RabbitmqQueue {
const queue = asRecord(raw)
return {
name: String(queue.name ?? ''),
vhost: String(queue.vhost ?? ''),
type: asStringOrNull(queue.type),
state: asStringOrNull(queue.state),
durable: asBooleanOrNull(queue.durable),
autoDelete: asBooleanOrNull(queue.auto_delete),
exclusive: asBooleanOrNull(queue.exclusive),
node: asStringOrNull(queue.node),
policy: asStringOrNull(queue.policy),
arguments: asRecord(queue.arguments),
messages: asNumberOrNull(queue.messages),
messagesReady: asNumberOrNull(queue.messages_ready),
messagesUnacknowledged: asNumberOrNull(queue.messages_unacknowledged),
consumers: asNumberOrNull(queue.consumers),
memory: asNumberOrNull(queue.memory),
}
}
export function projectExchange(raw: unknown): RabbitmqExchange {
const exchange = asRecord(raw)
return {
name: String(exchange.name ?? ''),
vhost: String(exchange.vhost ?? ''),
type: String(exchange.type ?? ''),
durable: exchange.durable === true,
autoDelete: exchange.auto_delete === true,
internal: exchange.internal === true,
arguments: asRecord(exchange.arguments),
}
}
export function projectBinding(raw: unknown): RabbitmqBinding {
const binding = asRecord(raw)
return {
source: String(binding.source ?? ''),
vhost: String(binding.vhost ?? ''),
destination: String(binding.destination ?? ''),
destinationType: String(binding.destination_type ?? ''),
routingKey: String(binding.routing_key ?? ''),
propertiesKey: String(binding.properties_key ?? ''),
arguments: asRecord(binding.arguments),
}
}
/**
* Projects a retrieved message. `payload_bytes` always reports the message's true size even
* when the broker truncated the returned payload, so comparing it against the requested limit
* makes truncation visible instead of silently handing back a short payload.
*/
export function projectMessage(raw: unknown, truncateLimit: number): RabbitmqMessage {
const message = asRecord(raw)
const payloadBytes = typeof message.payload_bytes === 'number' ? message.payload_bytes : 0
return {
truncated: payloadBytes > truncateLimit,
payload: String(message.payload ?? ''),
payloadBytes,
payloadEncoding: String(message.payload_encoding ?? ''),
exchange: String(message.exchange ?? ''),
routingKey: String(message.routing_key ?? ''),
redelivered: message.redelivered === true,
messageCount: typeof message.message_count === 'number' ? message.message_count : 0,
properties: asRecord(message.properties),
}
}
export const RABBITMQ_VHOST_OUTPUT_PROPERTIES = {
name: { type: 'string', description: 'Virtual host name' },
description: { type: 'string', description: 'Virtual host description' },
tags: { type: 'json', description: 'Tags applied to the virtual host' },
defaultQueueType: {
type: 'string',
description: 'Queue type used when a declaration does not specify one',
},
tracing: { type: 'boolean', description: 'Whether firehose tracing is enabled' },
messages: { type: 'number', description: 'Total messages across the virtual host' },
messagesReady: { type: 'number', description: 'Messages ready for delivery' },
messagesUnacknowledged: { type: 'number', description: 'Delivered but unacknowledged messages' },
clusterState: { type: 'json', description: 'Per-node state of the virtual host' },
} as const
export const RABBITMQ_CONNECTION_OUTPUT_PROPERTIES = {
name: { type: 'string', description: 'Connection name, formatted as peer -> broker' },
user: { type: 'string', description: 'Authenticated user' },
vhost: { type: 'string', description: 'Virtual host the connection is bound to' },
state: { type: 'string', description: 'Connection state, e.g. running, blocked, or flow' },
protocol: { type: 'string', description: 'Protocol and version, e.g. AMQP 0-9-1' },
node: { type: 'string', description: 'Cluster node holding the connection' },
channels: { type: 'number', description: 'Number of open channels on the connection' },
peerHost: { type: 'string', description: 'Client host address' },
peerPort: { type: 'number', description: 'Client port' },
connectedAt: { type: 'number', description: 'Connection start time in epoch milliseconds' },
ssl: { type: 'boolean', description: 'Whether the connection is TLS-encrypted' },
} as const
export const RABBITMQ_CHANNEL_OUTPUT_PROPERTIES = {
name: { type: 'string', description: 'Channel name, including its connection and number' },
number: { type: 'number', description: 'Channel number within its connection' },
user: { type: 'string', description: 'Authenticated user' },
vhost: { type: 'string', description: 'Virtual host the channel is bound to' },
node: { type: 'string', description: 'Cluster node holding the channel' },
state: { type: 'string', description: 'Channel state, e.g. running or flow' },
consumerCount: { type: 'number', description: 'Consumers registered on the channel' },
prefetchCount: { type: 'number', description: 'QoS prefetch limit, 0 when unlimited' },
messagesUnacknowledged: {
type: 'number',
description: 'Messages delivered on this channel awaiting acknowledgement',
},
confirm: { type: 'boolean', description: 'Whether publisher confirms are enabled' },
connectionName: { type: 'string', description: 'Name of the owning connection' },
} as const
export const RABBITMQ_CONSUMER_OUTPUT_PROPERTIES = {
consumerTag: { type: 'string', description: 'Consumer tag identifying the subscription' },
queue: { type: 'string', description: 'Queue the consumer reads from' },
vhost: { type: 'string', description: 'Virtual host the queue belongs to' },
ackRequired: {
type: 'boolean',
description: 'Whether the consumer must acknowledge messages explicitly',
},
active: {
type: 'boolean',
description: 'Whether the consumer is currently receiving messages',
},
activityStatus: {
type: 'string',
description: 'Why the consumer is or is not active, e.g. up or single_active',
},
exclusive: { type: 'boolean', description: 'Whether the consumer has exclusive queue access' },
prefetchCount: { type: 'number', description: 'QoS prefetch limit for the consumer' },
channelName: { type: 'string', description: 'Channel the consumer runs on' },
connectionName: { type: 'string', description: 'Connection the consumer runs on' },
} as const
export const RABBITMQ_NODE_OUTPUT_PROPERTIES = {
name: { type: 'string', description: 'Node name, e.g. rabbit@host' },
type: { type: 'string', description: 'Node type: disc or ram' },
running: { type: 'boolean', description: 'Whether the node is running' },
memUsed: { type: 'number', description: 'Memory used by the node in bytes' },
memLimit: { type: 'number', description: 'Memory high watermark in bytes' },
memAlarm: {
type: 'boolean',
description: 'Whether the memory alarm has fired, which blocks publishers',
},
diskFree: { type: 'number', description: 'Free disk space in bytes' },
diskFreeLimit: { type: 'number', description: 'Free disk space low watermark in bytes' },
diskFreeAlarm: {
type: 'boolean',
description: 'Whether the disk alarm has fired, which blocks publishers',
},
fdUsed: { type: 'number', description: 'File descriptors in use' },
fdTotal: { type: 'number', description: 'File descriptor limit' },
procUsed: { type: 'number', description: 'Erlang processes in use' },
procTotal: { type: 'number', description: 'Erlang process limit' },
uptime: { type: 'number', description: 'Node uptime in milliseconds' },
partitions: {
type: 'json',
description:
'Nodes this node considers network-partitioned from it. Non-empty means split brain',
},
beingDrained: {
type: 'boolean',
description: 'Whether the node is being drained for maintenance',
},
} as const
export const RABBITMQ_POLICY_OUTPUT_PROPERTIES = {
name: { type: 'string', description: 'Policy name' },
vhost: { type: 'string', description: 'Virtual host the policy applies in' },
pattern: { type: 'string', description: 'Regular expression matching queue or exchange names' },
applyTo: {
type: 'string',
description:
'What the policy applies to: all, queues, classic_queues, quorum_queues, streams, or exchanges',
},
priority: {
type: 'number',
description: 'Priority. Only the highest-priority matching policy applies to a given resource',
},
definition: { type: 'json', description: 'Policy definition keys applied to matching resources' },
} as const
export function projectVhost(raw: unknown): RabbitmqVhost {
const vhost = asRecord(raw)
return {
name: String(vhost.name ?? ''),
description: asStringOrNull(vhost.description),
tags: Array.isArray(vhost.tags) ? (vhost.tags as string[]) : [],
defaultQueueType: asStringOrNull(vhost.default_queue_type),
tracing: vhost.tracing === true,
messages: asNumberOrNull(vhost.messages),
messagesReady: asNumberOrNull(vhost.messages_ready),
messagesUnacknowledged: asNumberOrNull(vhost.messages_unacknowledged),
clusterState: asRecord(vhost.cluster_state),
}
}
export function projectConnection(raw: unknown): RabbitmqConnection {
const connection = asRecord(raw)
return {
name: String(connection.name ?? ''),
user: String(connection.user ?? ''),
vhost: String(connection.vhost ?? ''),
state: asStringOrNull(connection.state),
protocol: asStringOrNull(connection.protocol),
node: asStringOrNull(connection.node),
channels: asNumberOrNull(connection.channels),
peerHost: asStringOrNull(connection.peer_host),
peerPort: asNumberOrNull(connection.peer_port),
connectedAt: asNumberOrNull(connection.connected_at),
ssl: connection.ssl === true,
}
}
export function projectChannel(raw: unknown): RabbitmqChannel {
const channel = asRecord(raw)
const connectionDetails = asRecord(channel.connection_details)
return {
name: String(channel.name ?? ''),
number: asNumberOrNull(channel.number),
user: String(channel.user ?? ''),
vhost: String(channel.vhost ?? ''),
node: asStringOrNull(channel.node),
state: asStringOrNull(channel.state),
consumerCount: asNumberOrNull(channel.consumer_count),
prefetchCount: asNumberOrNull(channel.prefetch_count),
messagesUnacknowledged: asNumberOrNull(channel.messages_unacknowledged),
confirm: channel.confirm === true,
connectionName: asStringOrNull(connectionDetails.name),
}
}
export function projectConsumer(raw: unknown): RabbitmqConsumer {
const consumer = asRecord(raw)
const queue = asRecord(consumer.queue)
const channelDetails = asRecord(consumer.channel_details)
return {
consumerTag: String(consumer.consumer_tag ?? ''),
queue: String(queue.name ?? ''),
vhost: String(queue.vhost ?? ''),
ackRequired: consumer.ack_required === true,
active: consumer.active === true,
activityStatus: asStringOrNull(consumer.activity_status),
exclusive: consumer.exclusive === true,
prefetchCount: asNumberOrNull(consumer.prefetch_count),
channelName: asStringOrNull(channelDetails.name),
connectionName: asStringOrNull(channelDetails.connection_name),
}
}
export function projectNode(raw: unknown): RabbitmqNode {
const node = asRecord(raw)
return {
name: String(node.name ?? ''),
type: asStringOrNull(node.type),
running: node.running === true,
memUsed: asNumberOrNull(node.mem_used),
memLimit: asNumberOrNull(node.mem_limit),
memAlarm: node.mem_alarm === true,
diskFree: asNumberOrNull(node.disk_free),
diskFreeLimit: asNumberOrNull(node.disk_free_limit),
diskFreeAlarm: node.disk_free_alarm === true,
fdUsed: asNumberOrNull(node.fd_used),
fdTotal: asNumberOrNull(node.fd_total),
procUsed: asNumberOrNull(node.proc_used),
procTotal: asNumberOrNull(node.proc_total),
uptime: asNumberOrNull(node.uptime),
partitions: Array.isArray(node.partitions) ? (node.partitions as string[]) : [],
beingDrained: node.being_drained === true,
}
}
export function projectPolicy(raw: unknown): RabbitmqPolicy {
const policy = asRecord(raw)
return {
name: String(policy.name ?? ''),
vhost: String(policy.vhost ?? ''),
pattern: String(policy.pattern ?? ''),
applyTo: asStringOrNull(policy['apply-to']),
priority: asNumberOrNull(policy.priority),
definition: asRecord(policy.definition),
}
}
/** Columns requested from the connection and channel list endpoints. */
export const RABBITMQ_CONNECTION_COLUMNS = [
'name',
'user',
'vhost',
'state',
'protocol',
'node',
'channels',
'peer_host',
'peer_port',
'connected_at',
'ssl',
].join(',')
export const RABBITMQ_CHANNEL_COLUMNS = [
'name',
'number',
'user',
'vhost',
'node',
'state',
'consumer_count',
'prefetch_count',
'messages_unacknowledged',
'confirm',
'connection_details.name',
].join(',')
export const RABBITMQ_CONSUMER_COLUMNS = [
'consumer_tag',
'queue.name',
'queue.vhost',
'ack_required',
'active',
'activity_status',
'exclusive',
'prefetch_count',
'channel_details.name',
'channel_details.connection_name',
].join(',')
export const RABBITMQ_NODE_COLUMNS = [
'name',
'type',
'running',
'mem_used',
'mem_limit',
'mem_alarm',
'disk_free',
'disk_free_limit',
'disk_free_alarm',
'fd_used',
'fd_total',
'proc_used',
'proc_total',
'uptime',
'partitions',
'being_drained',
].join(',')
+52
View File
@@ -3143,6 +3143,33 @@ import {
quartrListTranscriptsTool,
} from '@/tools/quartr'
import { quiverImageToSvgTool, quiverListModelsTool, quiverTextToSvgTool } from '@/tools/quiver'
import {
rabbitmqCreateBindingTool,
rabbitmqCreateExchangeTool,
rabbitmqCreatePolicyTool,
rabbitmqCreateQueueTool,
rabbitmqDeleteBindingTool,
rabbitmqDeleteExchangeTool,
rabbitmqDeletePolicyTool,
rabbitmqDeleteQueueTool,
rabbitmqGetExchangeTool,
rabbitmqGetMessagesTool,
rabbitmqGetOverviewTool,
rabbitmqGetQueueTool,
rabbitmqHealthCheckTool,
rabbitmqListBindingsTool,
rabbitmqListChannelsTool,
rabbitmqListConnectionsTool,
rabbitmqListConsumersTool,
rabbitmqListExchangeBindingsTool,
rabbitmqListExchangesTool,
rabbitmqListNodesTool,
rabbitmqListPoliciesTool,
rabbitmqListQueuesTool,
rabbitmqListVhostsTool,
rabbitmqPublishMessageTool,
rabbitmqPurgeQueueTool,
} from '@/tools/rabbitmq'
import {
railwayCreateEnvironmentTool,
railwayCreateProjectTool,
@@ -8867,6 +8894,31 @@ export const tools: Record<string, ToolConfig> = {
qdrant_fetch_points: qdrantFetchTool,
qdrant_search_vector: qdrantSearchTool,
qdrant_upsert_points: qdrantUpsertTool,
rabbitmq_create_binding: rabbitmqCreateBindingTool,
rabbitmq_create_exchange: rabbitmqCreateExchangeTool,
rabbitmq_create_policy: rabbitmqCreatePolicyTool,
rabbitmq_create_queue: rabbitmqCreateQueueTool,
rabbitmq_delete_binding: rabbitmqDeleteBindingTool,
rabbitmq_delete_exchange: rabbitmqDeleteExchangeTool,
rabbitmq_delete_policy: rabbitmqDeletePolicyTool,
rabbitmq_delete_queue: rabbitmqDeleteQueueTool,
rabbitmq_get_exchange: rabbitmqGetExchangeTool,
rabbitmq_get_messages: rabbitmqGetMessagesTool,
rabbitmq_get_overview: rabbitmqGetOverviewTool,
rabbitmq_get_queue: rabbitmqGetQueueTool,
rabbitmq_health_check: rabbitmqHealthCheckTool,
rabbitmq_list_bindings: rabbitmqListBindingsTool,
rabbitmq_list_channels: rabbitmqListChannelsTool,
rabbitmq_list_connections: rabbitmqListConnectionsTool,
rabbitmq_list_consumers: rabbitmqListConsumersTool,
rabbitmq_list_exchange_bindings: rabbitmqListExchangeBindingsTool,
rabbitmq_list_exchanges: rabbitmqListExchangesTool,
rabbitmq_list_nodes: rabbitmqListNodesTool,
rabbitmq_list_policies: rabbitmqListPoliciesTool,
rabbitmq_list_queues: rabbitmqListQueuesTool,
rabbitmq_list_vhosts: rabbitmqListVhostsTool,
rabbitmq_publish_message: rabbitmqPublishMessageTool,
rabbitmq_purge_queue: rabbitmqPurgeQueueTool,
railway_create_environment: railwayCreateEnvironmentTool,
railway_create_project: railwayCreateProjectTool,
railway_create_service: railwayCreateServiceTool,