emqx-mcp-server

Instructions

Implementation Reference

• src/emqx_mcp_server/tools/emqx_message_tools.py:22-63 (handler)

  MCP tool handler for 'publish_mqtt_message'. Extracts and validates parameters from the request (topic, payload, qos, retain), then delegates to EMQXClient.publish_message() to perform the publish.

  @mcp.tool(name="publish_mqtt_message", 
            description="Publish an MQTT Message to Your EMQX Cluster on EMQX Cloud or Self-Managed Deployment")
  async def publish(request):
      """Handle publish message request
      
      Args:
          request: MCP request containing message data
              - topic: MQTT topic 
              - payload: Message content
              - qos: Quality of Service (0, 1, or 2)
              - retain: Whether to retain the message (true or false)
      
      Returns:
          MCPResponse: Response object with publish result
      """
      self.logger.info("Handling publish request")
      
      # Extract parameters from the request
      topic = request.get("topic")
      payload = request.get("payload")
      qos = request.get("qos", 0)  # Default QoS level is 0
      retain = request.get("retain", False)  # Default is not to retain
      
      # Validate required parameters before proceeding
      if not topic:
          self.logger.error("Missing required parameter: topic")
          return f'"error": "Missing required parameter: topic"'
      
      if payload is None:
          self.logger.error("Missing required parameter: payload")
          return f'"error": "Missing required parameter: payload"'
      
      # Publish message to EMQX using the client
      result = await self.emqx_client.publish_message(
          topic=topic,
          payload=payload,
          qos=qos,
          retain=retain
      )
      
      self.logger.info(f"Message published successfully to topic: {topic}")
      return result

• src/emqx_mcp_server/server.py:41-42 (registration)

  Instantiates EMQXMessageTools and calls its register_tools method on the FastMCP server instance, which defines and registers the 'publish_mqtt_message' tool using @mcp.tool decorator.

  emqx_message_tools = EMQXMessageTools(self.logger)
  emqx_message_tools.register_tools(self.mcp)

• src/emqx_mcp_server/emqx_client.py:56-87 (helper)

  Supporting utility in EMQXClient class that makes the actual HTTP POST request to the EMQX broker's /publish API endpoint to publish the MQTT message.

  async def publish_message(self, topic: str, payload: str, qos: int=0, retain: bool=False):
      """
      Publish a message to an MQTT topic.
      
      Uses the EMQX HTTP API to publish a message to a specific MQTT topic.
      
      Args:
          topic (str): The MQTT topic to publish to
          payload (str): The message payload to publish
          qos (int, optional): Quality of Service level (0, 1, or 2). Defaults to 0.
          retain (bool, optional): Whether to retain the message. Defaults to False.
          
      Returns:
          dict: Response from the EMQX API or error information
      """
      url = f"{self.api_url}/publish"
      data = {
          "topic": topic,
          "payload": payload,
          "qos": qos,
          "retain": retain
      }
      self.logger.info(f"Publishing message to topic {topic}")
      async with httpx.AsyncClient() as client:
          try:
              response = await client.post(url, headers=self._get_auth_header(), json=data, timeout=30)
              response.raise_for_status()
              return self._handle_response(response)
          except Exception as e:
              self.logger.error(f"Error publishing message: {str(e)}")
              return {"error": str(e)}