Add get_queue_info and delete_queue

Author: kenliao94

src/mcp_server_rabbitmq/handlers.py

@@ -21,3 +21,9 @@ def handle_list_queues(rabbitmq_admin: RabbitMQAdmin) -> List[str]:
 def handle_list_exchanges(rabbitmq_admin: RabbitMQAdmin) -> List[str]:
     result = rabbitmq_admin.list_exchanges()
     return [exchange['name'] for exchange in result]
+
+def handle_get_queue_info(rabbitmq_admin: RabbitMQAdmin, queue: str, vhost: str = "/") -> dict:
+    return rabbitmq_admin.get_queue_info(queue, vhost)
+
+def handle_delete_queue(rabbitmq_admin: RabbitMQAdmin, queue: str, vhost: str = "/") -> None:
+    rabbitmq_admin.delete_queue(queue, vhost)

src/mcp_server_rabbitmq/models.py

@@ -14,3 +14,11 @@ class ListQueues(BaseModel):
 
 class ListExchanges(BaseModel):
     pass
+
+class GetQueueInfo(BaseModel):
+    queue: Annotated[str, Field(description="The name of the queue to get info about")]
+    vhost: Annotated[str, Field(description="The virtual host where the queue exists")] = "/"
+
+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")] = "/"

src/mcp_server_rabbitmq/server.py

@@ -5,10 +5,17 @@
     Tool,
 )
 import ssl
-from .models import Enqueue, Fanout, ListQueues, ListExchanges
+from .models import Enqueue, Fanout, ListQueues, ListExchanges, GetQueueInfo, DeleteQueue
 from .logger import Logger, LOG_LEVEL
 from .connection import RabbitMQConnection, validate_rabbitmq_name
-from .handlers import handle_enqueue, handle_fanout, handle_list_queues, handle_list_exchanges
+from .handlers import (
+    handle_enqueue, 
+    handle_fanout, 
+    handle_list_queues, 
+    handle_list_exchanges,
+    handle_get_queue_info,
+    handle_delete_queue
+)
 from .admin import RabbitMQAdmin
 
 
@@ -48,6 +55,16 @@ async def list_tools() -> list[Tool]:
                 name="list_exchanges",
                 description="""List all the exchanges in the broker""",
                 inputSchema=ListExchanges.model_json_schema(),
+            ),
+            Tool(
+                name="get_queue_info",
+                description="""Get detailed information about a specific queue""",
+                inputSchema=GetQueueInfo.model_json_schema(),
+            ),
+            Tool(
+                name="delete_queue",
+                description="""Delete a specific queue""",
+                inputSchema=DeleteQueue.model_json_schema(),
             )
         ]
 
@@ -103,6 +120,28 @@ async def call_tool(
             except Exception as e:
                 logger.error(f"{e}")
                 return [TextContent(type="text", text=str("failed"))]
+        elif name == "get_queue_info":
+            try:
+                admin = RabbitMQAdmin(rabbitmq_host, api_port, username, password, use_tls)
+                queue = arguments["queue"]
+                vhost = arguments.get("vhost", "/")
+                validate_rabbitmq_name(queue, "Queue name")
+                result = handle_get_queue_info(admin, queue, vhost)
+                return [TextContent(type="text", text=str(result))]
+            except Exception as e:
+                logger.error(f"{e}")
+                return [TextContent(type="text", text=str("failed"))]
+        elif name == "delete_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_delete_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"))]
         raise ValueError(f"Tool not found: {name}")
 
     options = server.create_initialization_options()