emqx-mcp-server

Instructions

Implementation Reference

• src/emqx_mcp_server/tools/emqx_client_tools.py:70-94 (handler)

  The MCP tool handler function for get_mqtt_client: extracts clientid from request, validates it, calls EMQXClient.get_client_info(clientid) and returns the result.

  @mcp.tool(name="get_mqtt_client", 
            description="Get detailed information about a specific MQTT client by client ID")
  async def get_client_info(request):
      """Handle get client info request
      
      Args:
          request: MCP request containing client identifier
              - clientid: Client ID (required) - The unique identifier of the client to retrieve
  
      Returns:
          MCPResponse: Response object with detailed client information
      """
      self.logger.info("Handling get client info request")
      
      # Extract required client ID parameter
      clientid = request.get("clientid")
      if not clientid:
          self.logger.error("Client ID is required but was not provided")
          return {"error": "Client ID is required"}
      
      # Get client information from EMQX
      result = await self.emqx_client.get_client_info(clientid)
      
      self.logger.info(f"Client info for '{clientid}' retrieved successfully")
      return result 

• src/emqx_mcp_server/emqx_client.py:133-162 (helper)

  Core helper method that performs the HTTP GET request to the EMQX API (/clients/{clientid}) to retrieve detailed client information, handles auth and errors.

  async def get_client_info(self, clientid: str):
      """
      Get detailed information about a specific MQTT client by client ID.
      
      Uses the EMQX HTTP API to retrieve detailed information about a specific
      client identified by its client ID.
      
      Args:
          clientid (str): The unique identifier of the client to retrieve
          
      Returns:
          dict: Response from the EMQX API containing client data or error information
      """
      url = f"{self.api_url}/clients/{clientid}"
      
      self.logger.info(f"Retrieving information for client ID: {clientid}")
      
      async with httpx.AsyncClient() as client:
          try:
              response = await client.get(
                  url,
                  headers=self._get_auth_header(),
                  timeout=30
              )
              response.raise_for_status()
              return self._handle_response(response)
          except Exception as e:
              self.logger.error(f"Error retrieving client info for {clientid}: {str(e)}")
              return {"error": str(e)}

• src/emqx_mcp_server/server.py:45-48 (registration)

  Registers the EMQXClientTools instance (which defines and registers the get_mqtt_client tool via @mcp.tool decorator) with the FastMCP server.

  # Register client tools
  emqx_client_tools = EMQXClientTools(self.logger)
  emqx_client_tools.register_tools(self.mcp)
  self.logger.info("EMQX client tools registered")