get_mqtt_client
Retrieve detailed information about a specific MQTT client using its client ID for monitoring and management purposes.
Instructions
Get detailed information about a specific MQTT client by client ID
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| request | Yes |
Implementation Reference
- 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 - 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")