get_mqtt_client
Retrieve detailed information about a specific MQTT client using its client ID to monitor and manage connections on the EMQX MQTT broker.
Instructions
Get detailed information about a specific MQTT client by client ID
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| request | Yes |
Implementation Reference
- MCP tool handler function for 'get_mqtt_client'. Handles the MCP request, validates clientid parameter, delegates to EMQXClient.get_client_info, 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
- Core helper method implementing the HTTP API call to EMQX /clients/{clientid} endpoint to fetch detailed client information. Includes authentication, error handling, and response processing.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 with the MCP server by calling register_tools(self.mcp), which defines and registers the 'get_mqtt_client' tool via decorator.# Register client tools emqx_client_tools = EMQXClientTools(self.logger) emqx_client_tools.register_tools(self.mcp) self.logger.info("EMQX client tools registered")