Add fanout tool

Author: kenliao94

src/mcp_server_rabbitmq/models.py

@@ -3,4 +3,8 @@
 
 class Enqueue(BaseModel):
     message: Annotated[str, Field(description="The message to publish")]
-    queue: Annotated[str, Field(description="The name of the queue")]
+    queue: Annotated[str, Field(description="The name of the queue")]
+
+class Fanout(BaseModel):
+    message: Annotated[str, Field(description="The message to publish")]
+    exchange: Annotated[str, Field(description="The name of the exchange")]

src/mcp_server_rabbitmq/server.py

@@ -6,7 +6,7 @@
 )
 import pika
 import ssl
-from .models import Enqueue
+from .models import Enqueue, Fanout
 from .logger import Logger, LOG_LEVEL
 
 
@@ -39,6 +39,11 @@ async def list_tools() -> list[Tool]:
                 name="enqueue",
                 description="""Enqueue a message to a queue hosted on RabbitMQ""",
                 inputSchema=Enqueue.model_json_schema(),
+            ),
+            Tool(
+                name="fanout",
+                description="""Publish a message to an exchange with fanout type""",
+                inputSchema=Fanout.model_json_schema(),
             )
         ]
 
@@ -51,7 +56,6 @@ async def call_tool(
             logger.debug("Executing enqueue tool")
             message = arguments["message"]
             queue = arguments["queue"]
-            # Send to RabbitMQ host
             try:
                 connection = pika.BlockingConnection(parameters)
                 channel = connection.channel()
@@ -61,6 +65,19 @@ async def call_tool(
             except Exception as e:
                 logger.error(f"{e}")
                 return [TextContent(type="text", text=str("failed"))]
+        elif name == "fanout":
+            logger.debug("Executing fanout tool")
+            exchange = arguments["exchange"]
+            message = arguments["message"]
+            try:
+                connection = pika.BlockingConnection(parameters)
+                channel = connection.channel()
+                channel.exchange_declare(exchange=exchange, exchange_type="fanout")
+                channel.basic_publish(exchange=exchange, routing_key="", body=message)
+                return [TextContent(type="text", text=str("suceeded"))]
+            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()