list_mqtt_clients
Retrieve a list of all MQTT clients currently connected to your EMQX cluster for monitoring and management purposes.
Instructions
List MQTT clients connected to your EMQX Cluster
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| request | Yes |
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):
- 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
- Helper method in EMQXClient class that performs the actual HTTP GET request to the EMQX /clients API endpoint to retrieve the list of MQTT clientsasync 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)}