Add "handle_purge_queue" "handle_delete_exchange" and "handle_get_exchange_info"

Author: kenliao94

src/mcp_server_rabbitmq/handlers.py

@@ -27,3 +27,12 @@ def handle_get_queue_info(rabbitmq_admin: RabbitMQAdmin, queue: str, vhost: str
 
 def handle_delete_queue(rabbitmq_admin: RabbitMQAdmin, queue: str, vhost: str = "/") -> None:
     rabbitmq_admin.delete_queue(queue, vhost)
+
+def handle_purge_queue(rabbitmq_admin: RabbitMQAdmin, queue: str, vhost: str = "/") -> None:
+    rabbitmq_admin.purge_queue(queue, vhost)
+
+def handle_delete_exchange(rabbitmq_admin: RabbitMQAdmin, exchange: str, vhost: str = "/") -> None:
+    rabbitmq_admin.delete_exchange(exchange, vhost)
+
+def handle_get_exchange_info(rabbitmq_admin: RabbitMQAdmin, exchange: str, vhost: str = "/") -> dict:
+    return rabbitmq_admin.get_exchange_info(exchange, vhost)

src/mcp_server_rabbitmq/models.py

@@ -22,3 +22,15 @@ class GetQueueInfo(BaseModel):
 class DeleteQueue(BaseModel):
     queue: Annotated[str, Field(description="The name of the queue to delete")]
     vhost: Annotated[str, Field(description="The virtual host where the queue exists")] = "/"
+
+class PurgeQueue(BaseModel):
+    queue: Annotated[str, Field(description="The name of the queue to purge")]
+    vhost: Annotated[str, Field(description="The virtual host where the queue exists")] = "/"
+
+class DeleteExchange(BaseModel):
+    exchange: Annotated[str, Field(description="The name of the exchange to delete")]
+    vhost: Annotated[str, Field(description="The virtual host where the exchange exists")] = "/"
+
+class GetExchangeInfo(BaseModel):
+    exchange: Annotated[str, Field(description="The name of the exchange to get info about")]
+    vhost: Annotated[str, Field(description="The virtual host where the exchange exists")] = "/"

src/mcp_server_rabbitmq/server.py

@@ -5,7 +5,17 @@
     Tool,
 )
 import ssl
-from .models import Enqueue, Fanout, ListQueues, ListExchanges, GetQueueInfo, DeleteQueue
+from .models import (
+    Enqueue, 
+    Fanout, 
+    ListQueues, 
+    ListExchanges, 
+    GetQueueInfo, 
+    DeleteQueue,
+    PurgeQueue,
+    DeleteExchange,
+    GetExchangeInfo
+)
 from .logger import Logger, LOG_LEVEL
 from .connection import RabbitMQConnection, validate_rabbitmq_name
 from .handlers import (
@@ -14,7 +24,10 @@
     handle_list_queues, 
     handle_list_exchanges,
     handle_get_queue_info,
-    handle_delete_queue
+    handle_delete_queue,
+    handle_purge_queue,
+    handle_delete_exchange,
+    handle_get_exchange_info
 )
 from .admin import RabbitMQAdmin
 
@@ -65,6 +78,21 @@ async def list_tools() -> list[Tool]:
                 name="delete_queue",
                 description="""Delete a specific queue""",
                 inputSchema=DeleteQueue.model_json_schema(),
+            ),
+            Tool(
+                name="purge_queue",
+                description="""Remove all messages from a specific queue""",
+                inputSchema=PurgeQueue.model_json_schema(),
+            ),
+            Tool(
+                name="delete_exchange",
+                description="""Delete a specific exchange""",
+                inputSchema=DeleteExchange.model_json_schema(),
+            ),
+            Tool(
+                name="get_exchange_info",
+                description="""Get detailed information about a specific exchange""",
+                inputSchema=GetExchangeInfo.model_json_schema(),
             )
         ]
 
@@ -142,6 +170,39 @@ async def call_tool(
             except Exception as e:
                 logger.error(f"{e}")
                 return [TextContent(type="text", text=str("failed"))]
+        elif name == "purge_queue":
+            try:
+                admin = RabbitMQAdmin(rabbitmq_host, api_port, username, password, use_tls)
+                queue = arguments["queue"]
+                vhost = arguments.get("vhost", "/")
+                validate_rabbitmq_name(queue, "Queue name")
+                handle_purge_queue(admin, queue, vhost)
+                return [TextContent(type="text", text=str("succeeded"))]
+            except Exception as e:
+                logger.error(f"{e}")
+                return [TextContent(type="text", text=str("failed"))]
+        elif name == "delete_exchange":
+            try:
+                admin = RabbitMQAdmin(rabbitmq_host, api_port, username, password, use_tls)
+                exchange = arguments["exchange"]
+                vhost = arguments.get("vhost", "/")
+                validate_rabbitmq_name(exchange, "Exchange name")
+                handle_delete_exchange(admin, exchange, vhost)
+                return [TextContent(type="text", text=str("succeeded"))]
+            except Exception as e:
+                logger.error(f"{e}")
+                return [TextContent(type="text", text=str("failed"))]
+        elif name == "get_exchange_info":
+            try:
+                admin = RabbitMQAdmin(rabbitmq_host, api_port, username, password, use_tls)
+                exchange = arguments["exchange"]
+                vhost = arguments.get("vhost", "/")
+                validate_rabbitmq_name(exchange, "Exchange name")
+                result = handle_get_exchange_info(admin, exchange, vhost)
+                return [TextContent(type="text", text=str(result))]
+            except Exception as e:
+                logger.error(f"{e}")
+                return [TextContent(type="text", text=str("failed"))]
         raise ValueError(f"Tool not found: {name}")
 
     options = server.create_initialization_options()