list_mqtt_clients

Instructions

Implementation Reference

• src/emqx_mcp_server/tools/emqx_client_tools.py:22-24 (registration)

  Registration of the 'list_mqtt_clients' tool using the @mcp.tool decorator inside register_tools method

  @mcp.tool(name="list_mqtt_clients", 
            description="List MQTT clients connected to your EMQX Cluster")
  async def list_clients(request):

• src/emqx_mcp_server/tools/emqx_client_tools.py:25-69 (handler)

  The tool handler function that parses MCP request parameters and calls the EMQX client to list MQTT clients

  """Handle list clients request
  
  Args:
      request: MCP request containing filter parameters
          - page: Page number (default: 1)
          - limit: Results per page, max 10000 (default: 10)
          - node: Node name
          - clientid: Client ID
          - username: Username
          - ip_address: Client IP address
          - conn_state: Connection state
          - clean_start: Clean start flag
          - proto_ver: Protocol version
          - like_clientid: Fuzzy search by client ID pattern
          - like_username: Fuzzy search by username pattern
          - like_ip_address: Fuzzy search by IP address pattern
  
  Returns:
      MCPResponse: Response object with list of clients
  """
  self.logger.info("Handling list clients request")
  
  # Extract optional parameters from the request with defaults
  params = {
      "page": request.get("page", 1),
      "limit": request.get("limit", 100)
  }
  
  # Optional parameters to include if present
  optional_params = [
      "node", "clientid", "username", "ip_address", "conn_state", 
      "clean_start", "proto_ver", "like_clientid", "like_username", 
      "like_ip_address"
  ]
  
  for param in optional_params:
      if param in request:
          params[param] = request.get(param)
  
  # Get list of clients from EMQX
  result = await self.emqx_client.list_clients(params)
  
  self.logger.info("Client list retrieved successfully")
  return result 

• src/emqx_mcp_server/emqx_client.py:88-132 (helper)

  Helper method in EMQXClient class that performs the actual HTTP GET request to the EMQX /clients API endpoint to retrieve the list of MQTT clients

  async def list_clients(self, params=None):
      """
      Get a list of connected MQTT clients.
  
      Uses the EMQX HTTP API to retrieve information about connected clients.
      
      Args:
          params (dict, optional): Query parameters to filter results:
              - page: Page number (default: 1)
              - limit: Results per page, max 10000 (default: 10)
              - clientid: Client ID
              - username: Username
              - ip_address: Client IP address
              - conn_state: Connection state
              - clean_start: Clean start flag
              - proto_ver: Protocol version
              - like_clientid: Fuzzy search by client ID pattern
              - like_username: Fuzzy search by username pattern
              - like_ip_address: Fuzzy search by IP address pattern
  
      Returns:
          dict: Response from the EMQX API containing client data or error information
      """
      url = f"{self.api_url}/clients"
  
      # Default params if none provided
      if params is None:
          params = {"page": 1, "limit": 10}
  
      self.logger.info("Retrieving list of MQTT clients")
  
      async with httpx.AsyncClient() as client:
          try:
              response = await client.get(
                  url,
                  headers=self._get_auth_header(),
                  params=params,
                  timeout=30
              )
              response.raise_for_status()
              return self._handle_response(response)
          except Exception as e:
              self.logger.error(f"Error retrieving clients: {str(e)}")
              return {"error": str(e)}